Refactor podman

CHANGELOG.md

@@ -17,6 +17,7 @@
 - `prometheusreceiver`: Add `target_info` labels to resource attributes. (#11034)
 - `saphanareceiver`: Fix component memory query, add better error handling (#11507)
 - `sapmexporter`: Add config option to log responses from Splunk APM. (#11425)
+- `podmanreceiver`: Fetch containers stats one by one and add container image as metric attribute (#10421)
 
 ### 🧰 Bug fixes 🧰
 

receiver/podmanreceiver/go.mod

@@ -3,6 +3,8 @@ module github.com/open-telemetry/opentelemetry-collector-contrib/receiver/podman
 go 1.17
 
 require (
+
+	go.uber.org/multierr v1.8.0
 	github.com/stretchr/testify v1.7.5
 	go.opentelemetry.io/collector v0.54.0
 	go.opentelemetry.io/collector/pdata v0.54.0
@@ -32,7 +34,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
 	}
@@ -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,50 @@ func (c *podmanClient) ping(ctx context.Context) error {
 	}
 	return nil
 }
+
+// events returns a stream of events. It's up to the caller to close the stream by canceling the context.
+func (c *libpodClient) events(ctx context.Context, options url.Values) (<-chan event, <-chan error) {
+	events := make(chan event)
+	errs := make(chan error, 1)
+
+	started := make(chan struct{})
+	go func() {
+		defer close(errs)
+
+		resp, err := c.request(ctx, "/events", options)
+		if err != nil {
+			close(started)
+			errs <- err
+			return
+		}
+		defer resp.Body.Close()
+
+		dec := json.NewDecoder(resp.Body)
+		close(started)
+		for {
+			var e event
+			select {
+			case <-ctx.Done():
+				errs <- ctx.Err()
+				return
+			default:
+				err := dec.Decode(&e)
+				if err != nil {
+					errs <- err
+					return
+				}
+
+				select {
+				case events <- e:
+				case <-ctx.Done():
+					errs <- ctx.Err()
+					return
+				}
+			}
+		}
+	}()
+
+	<-started
+
+	return events, errs
+}