Refactor podman

CHANGELOG.md

@@ -24,6 +24,7 @@
 - `tailsamplingprocessor`: Add support for string invert matching to `and` policy (#9553)
 - `mezemoexporter`: Add user agent string to outgoing HTTP requests (#10470)
 - `transformprocessor`: Add functions for conversion of scalar metric types (`gauge_to_sum` and `sum_to_gauge`) (#10255)
+- `podmanreceiver`: Fetch containers stats one by one and add container image as metric attribute (#10421)
 
 ### 🧰 Bug fixes 🧰
 

receiver/podmanreceiver/go.mod

@@ -7,6 +7,7 @@ require (
 	go.opentelemetry.io/collector v0.52.0
 	go.opentelemetry.io/collector/pdata v0.52.0
 	go.opentelemetry.io/collector/semconv v0.52.0
+	go.uber.org/multierr v1.8.0
 	go.uber.org/zap v1.21.0
 	golang.org/x/crypto v0.0.0-20220507011949-2cf3adece122
 )
@@ -32,7 +33,6 @@ require (
 	go.opentelemetry.io/otel/metric v0.30.0 // indirect
 	go.opentelemetry.io/otel/trace v1.7.0 // indirect
 	go.uber.org/atomic v1.9.0 // indirect
-	go.uber.org/multierr v1.8.0 // indirect
 	golang.org/x/net v0.0.0-20220225172249-27dd8689420f // indirect
 	golang.org/x/sys v0.0.0-20220412211240-33da011f77ad // indirect
 	golang.org/x/text v0.3.7 // indirect

receiver/podmanreceiver/podman_client.go → receiver/podmanreceiver/libpod_client.go

@@ -25,7 +25,6 @@ import (
 	"io/ioutil"
 	"net/http"
 	"net/url"
-	"time"
 
 	"go.uber.org/zap"
 )
@@ -34,68 +33,24 @@ var (
 	errNoStatsFound = fmt.Errorf("No stats found")
 )
 
-type containerStats struct {
-	AvgCPU        float64
-	ContainerID   string
-	Name          string
-	PerCPU        []uint64
-	CPU           float64
-	CPUNano       uint64
-	CPUSystemNano uint64
-	DataPoints    int64
-	SystemNano    uint64
-	MemUsage      uint64
-	MemLimit      uint64
-	MemPerc       float64
-	NetInput      uint64
-	NetOutput     uint64
-	BlockInput    uint64
-	BlockOutput   uint64
-	PIDs          uint64
-	UpTime        time.Duration
-	Duration      uint64
-}
-
-type containerStatsReportError struct {
-	Cause    string
-	Message  string
-	Response int64
-}
-
-type containerStatsReport struct {
-	Error containerStatsReportError
-	Stats []containerStats
-}
-
-type clientFactory func(logger *zap.Logger, cfg *Config) (client, error)
-
-type client interface {
-	ping(context.Context) error
-	stats(context.Context) ([]containerStats, error)
-}
-
-type podmanClient struct {
+type libpodClient struct {
 	conn     *http.Client
 	endpoint string
-
-	// The maximum amount of time to wait for Podman API responses
-	timeout time.Duration
 }
 
-func newPodmanClient(logger *zap.Logger, cfg *Config) (client, error) {
+func newLibpodClient(logger *zap.Logger, cfg *Config) (PodmanClient, error) {
 	connection, err := newPodmanConnection(logger, cfg.Endpoint, cfg.SSHKey, cfg.SSHPassphrase)
 	if err != nil {
 		return nil, err
 	}
-	c := &podmanClient{
+	c := &libpodClient{
 		conn:     connection,
 		endpoint: fmt.Sprintf("http://d/v%s/libpod", cfg.APIVersion),
-		timeout:  cfg.Timeout,
 	}
 	return c, nil
 }
 
-func (c *podmanClient) request(ctx context.Context, path string, params url.Values) (*http.Response, error) {
+func (c *libpodClient) request(ctx context.Context, path string, params url.Values) (*http.Response, error) {
 	req, err := http.NewRequestWithContext(ctx, "GET", c.endpoint+path, nil)
 	if err != nil {
 		return nil, err
@@ -107,13 +62,8 @@ func (c *podmanClient) request(ctx context.Context, path string, params url.Valu
 	return c.conn.Do(req)
 }
 
-func (c *podmanClient) stats(ctx context.Context) ([]containerStats, error) {
-	params := url.Values{}
-	params.Add("stream", "false")
-
-	statsCtx, cancel := context.WithTimeout(ctx, c.timeout)
-	defer cancel()
-	resp, err := c.request(statsCtx, "/containers/stats", params)
+func (c *libpodClient) stats(ctx context.Context, options url.Values) ([]ContainerStats, error) {
+	resp, err := c.request(ctx, "/containers/stats", options)
 	if err != nil {
 		return nil, err
 	}
@@ -124,7 +74,7 @@ func (c *podmanClient) stats(ctx context.Context) ([]containerStats, error) {
 		return nil, err
 	}
 
-	report := &containerStatsReport{}
+	report := &ContainerStatsReport{}
 	err = json.Unmarshal(bytes, report)
 	if err != nil {
 		return nil, err
@@ -138,10 +88,28 @@ func (c *podmanClient) stats(ctx context.Context) ([]containerStats, error) {
 	return report.Stats, nil
 }
 
-func (c *podmanClient) ping(ctx context.Context) error {
-	pingCtx, cancel := context.WithTimeout(ctx, c.timeout)
-	defer cancel()
-	resp, err := c.request(pingCtx, "/_ping", nil)
+func (c *libpodClient) list(ctx context.Context, options url.Values) ([]Container, error) {
+	resp, err := c.request(ctx, "/containers/json", options)
+	if err != nil {
+		return nil, err
+	}
+	defer resp.Body.Close()
+
+	bytes, err := ioutil.ReadAll(resp.Body)
+	if err != nil {
+		return nil, err
+	}
+
+	var report []Container
+	err = json.Unmarshal(bytes, &report)
+	if err != nil {
+		return nil, err
+	}
+	return report, nil
+}
+
+func (c *libpodClient) ping(ctx context.Context) error {
+	resp, err := c.request(ctx, "/_ping", nil)
 	if err != nil {
 		return err
 	}
@@ -151,3 +119,30 @@ func (c *podmanClient) ping(ctx context.Context) error {
 	}
 	return nil
 }
+
+func (c *libpodClient) events(ctx context.Context, options url.Values) (<-chan Event, <-chan error) {
+	events := make(chan Event)
+	errs := make(chan error)
+	go func() {
+		resp, err := c.request(ctx, "/events", options)
+		if err != nil {
+			errs <- err
+		}
+		dec := json.NewDecoder(resp.Body)
+		for {
+			var e Event
+			select {
+			case <-ctx.Done():
+				return
+			default:
+				err := dec.Decode(&e)
+				if err != nil {
+					errs <- err
+				} else {
+					events <- e
+				}
+			}
+		}
+	}()
+	return events, errs
+}

receiver/podmanreceiver/podman_client_test.go → receiver/podmanreceiver/libpod_client_test.go

@@ -30,7 +30,6 @@ import (
 	"time"
 
 	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/require"
 	"go.uber.org/zap"
 )
 
@@ -50,38 +49,6 @@ func tmpSock(t *testing.T) (net.Listener, string) {
 	return listener, addr
 }
 
-func TestWatchingTimeouts(t *testing.T) {
-	listener, addr := tmpSock(t)
-	defer listener.Close()
-	defer os.Remove(addr)
-
-	config := &Config{
-		Endpoint: fmt.Sprintf("unix://%s", addr),
-		Timeout:  50 * time.Millisecond,
-	}
-
-	cli, err := newPodmanClient(zap.NewNop(), config)
-	assert.NotNil(t, cli)
-	assert.Nil(t, err)
-
-	expectedError := "context deadline exceeded"
-
-	shouldHaveTaken := time.Now().Add(100 * time.Millisecond).UnixNano()
-
-	err = cli.ping(context.Background())
-	require.Error(t, err)
-
-	containers, err := cli.stats(context.Background())
-	require.Error(t, err)
-	assert.Contains(t, err.Error(), expectedError)
-	assert.Nil(t, containers)
-
-	assert.GreaterOrEqual(
-		t, time.Now().UnixNano(), shouldHaveTaken,
-		"Client timeouts don't appear to have been exercised.",
-	)
-}
-
 func TestStats(t *testing.T) {
 	// stats sample
 	statsExample := `{"Error":null,"Stats":[{"AvgCPU":42.04781177856639,"ContainerID":"e6af5805edae6c950003abd5451808b277b67077e400f0a6f69d01af116ef014","Name":"charming_sutherland","PerCPU":null,"CPU":42.04781177856639,"CPUNano":309165846000,"CPUSystemNano":54515674,"SystemNano":1650912926385978706,"MemUsage":27717632,"MemLimit":7942234112,"MemPerc":0.34899036730888044,"NetInput":430,"NetOutput":330,"BlockInput":0,"BlockOutput":0,"PIDs":118,"UpTime":309165846000,"Duration":309165846000}]}`
@@ -109,11 +76,11 @@ func TestStats(t *testing.T) {
 		Timeout: 5 * time.Second,
 	}
 
-	cli, err := newPodmanClient(zap.NewNop(), config)
+	cli, err := newLibpodClient(zap.NewNop(), config)
 	assert.NotNil(t, cli)
 	assert.Nil(t, err)
 
-	expectedStats := containerStats{
+	expectedStats := ContainerStats{
 		AvgCPU:        42.04781177856639,
 		ContainerID:   "e6af5805edae6c950003abd5451808b277b67077e400f0a6f69d01af116ef014",
 		Name:          "charming_sutherland",
@@ -134,7 +101,7 @@ func TestStats(t *testing.T) {
 		Duration:      309165846000,
 	}
 
-	stats, err := cli.stats(context.Background())
+	stats, err := cli.stats(context.Background(), nil)
 	assert.NoError(t, err)
 	assert.Equal(t, expectedStats, stats[0])
 }
@@ -167,11 +134,75 @@ func TestStatsError(t *testing.T) {
 		Timeout: 5 * time.Second,
 	}
 
-	cli, err := newPodmanClient(zap.NewNop(), config)
+	cli, err := newLibpodClient(zap.NewNop(), config)
 	assert.NotNil(t, cli)
 	assert.Nil(t, err)
 
-	stats, err := cli.stats(context.Background())
+	stats, err := cli.stats(context.Background(), nil)
 	assert.Nil(t, stats)
 	assert.EqualError(t, err, errNoStatsFound.Error())
 }
+
+func TestList(t *testing.T) {
+	// list sample
+	listExample := `[{"AutoRemove":false,"Command":["nginx","-g","daemon off;"],"Created":"2022-05-28T11:25:35.999277074+02:00","CreatedAt":"","Exited":false,"ExitedAt":-62135596800,"ExitCode":0,"Id":"aa3e2040dee22a369d2c8f0b712a5ff045e8f1ce47f5e943426ce6664e3ef379","Image":"library/nginxy:latest","ImageID":"12766a6745eea133de9fdcd03ff720fa971fdaf21113d4bc72b417c123b15619","IsInfra":false,"Labels":{"maintainer":"someone"},"Mounts":[],"Names":["sharp_curran"],"Namespaces":{},"Networks":[],"Pid":7892,"Pod":"","PodName":"","Ports":null,"Size":null,"StartedAt":1653729936,"State":"running","Status":""}]`
+
+	listener, addr := tmpSock(t)
+	defer listener.Close()
+	defer os.Remove(addr)
+
+	srv := httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if strings.Contains(r.URL.Path, "/containers/json") {
+			_, err := w.Write([]byte(listExample))
+			assert.NoError(t, err)
+		} else {
+			_, err := w.Write([]byte{})
+			assert.NoError(t, err)
+		}
+	}))
+	srv.Listener = listener
+	srv.Start()
+	defer srv.Close()
+
+	config := &Config{
+		Endpoint: fmt.Sprintf("unix://%s", addr),
+		// default timeout
+		Timeout: 5 * time.Second,
+	}
+
+	cli, err := newLibpodClient(zap.NewNop(), config)
+	assert.NotNil(t, cli)
+	assert.Nil(t, err)
+
+	expectedContainer := Container{
+
+		AutoRemove: false,
+		Command:    []string{"nginx", "-g", "daemon off;"},
+		Created:    "2022-05-28T11:25:35.999277074+02:00",
+		CreatedAt:  "",
+		Exited:     false,
+		ExitedAt:   -62135596800,
+		ExitCode:   0,
+		ID:         "aa3e2040dee22a369d2c8f0b712a5ff045e8f1ce47f5e943426ce6664e3ef379",
+		Image:      "library/nginxy:latest",
+		ImageID:    "12766a6745eea133de9fdcd03ff720fa971fdaf21113d4bc72b417c123b15619",
+		IsInfra:    false,
+		Labels:     map[string]string{"maintainer": "someone"},
+		Mounts:     []string{},
+		Names:      []string{"sharp_curran"},
+		Namespaces: map[string]string{},
+		Networks:   []string{},
+		Pid:        7892,
+		Pod:        "",
+		PodName:    "",
+		Ports:      nil,
+		Size:       nil,
+		StartedAt:  1653729936,
+		State:      "running",
+		Status:     "",
+	}
+
+	containers, err := cli.list(context.Background(), nil)
+	assert.NoError(t, err)
+	assert.Equal(t, expectedContainer, containers[0])
+}