[receiver/podman] Add timeout config option

CHANGELOG.md

@@ -11,6 +11,7 @@
 - `jaegerremotesamplingextension`: Add local and remote sampling stores (#8818)
 - `attributesprocessor`: Add support to filter on log body (#8996)
 - `prometheusremotewriteexporter`: Translate resource attributes to the target info metric (#8493)
+- `podmanreceiver`: Add API timeout configuration option (#9014)
 - `cmd/mdatagen`: Add `sem_conv_version` field to metadata.yaml that is used to set metrics SchemaURL (#9010)
 
 ### 🛑 Breaking changes 🛑

receiver/podmanreceiver/README.md

@@ -19,13 +19,15 @@ The following settings are required:
 The following settings are optional:
 
 - `collection_interval` (default = `10s`): The interval at which to gather container stats.
+- `timeout` (default = `5s`): The maximum amount of time to wait for Podman API responses.
 
 Example:
 
 ```yaml
 receivers:
   podman_stats:
     endpoint: unix://run/podman/podman.sock
+    timeout: 10s
     collection_interval: 10s
 ```
 
@@ -81,4 +83,4 @@ Recommended build tags to use when including this receiver in your build:
 
 - `containers_image_openpgp`
 - `exclude_graphdriver_btrfs`
-- `exclude_graphdriver_devicemapper`
+- `exclude_graphdriver_devicemapper`

receiver/podmanreceiver/config.go

@@ -16,6 +16,7 @@ package podmanreceiver // import "github.com/open-telemetry/opentelemetry-collec
 
 import (
 	"errors"
+	"time"
 
 	"go.opentelemetry.io/collector/config"
 	"go.opentelemetry.io/collector/receiver/scraperhelper"
@@ -27,7 +28,11 @@ type Config struct {
 	scraperhelper.ScraperControllerSettings `mapstructure:",squash"`
 
 	// The URL of the podman server.  Default is "unix:///run/podman/podman.sock"
-	Endpoint      string `mapstructure:"endpoint"`
+	Endpoint string `mapstructure:"endpoint"`
+
+	// The maximum amount of time to wait for Podman API responses.  Default is 5s
+	Timeout time.Duration `mapstructure:"timeout"`
+
 	APIVersion    string `mapstructure:"api_version"`
 	SSHKey        string `mapstructure:"ssh_key"`
 	SSHPassphrase string `mapstructure:"ssh_passphrase"`

receiver/podmanreceiver/config_test.go

@@ -45,9 +45,11 @@ func TestLoadConfig(t *testing.T) {
 	assert.Equal(t, "podman_stats", dcfg.ID().String())
 	assert.Equal(t, "unix:///run/podman/podman.sock", dcfg.Endpoint)
 	assert.Equal(t, 10*time.Second, dcfg.CollectionInterval)
+	assert.Equal(t, 5*time.Second, dcfg.Timeout)
 
 	ascfg := cfg.Receivers[config.NewComponentIDWithName(typeStr, "all")].(*Config)
 	assert.Equal(t, "podman_stats/all", ascfg.ID().String())
 	assert.Equal(t, "http://example.com/", ascfg.Endpoint)
 	assert.Equal(t, 2*time.Second, ascfg.CollectionInterval)
+	assert.Equal(t, 20*time.Second, ascfg.Timeout)
 }

receiver/podmanreceiver/factory.go

@@ -43,6 +43,7 @@ func createDefaultConfig() *Config {
 			CollectionInterval: 10 * time.Second,
 		},
 		Endpoint:   "unix:///run/podman/podman.sock",
+		Timeout:    5 * time.Second,
 		APIVersion: defaultAPIVersion,
 	}
 }

receiver/podmanreceiver/podman_client.go

@@ -60,12 +60,16 @@ type containerStatsReport struct {
 type clientFactory func(logger *zap.Logger, cfg *Config) (client, error)
 
 type client interface {
-	stats() ([]containerStats, error)
+	ping(context.Context) error
+	stats(context.Context) ([]containerStats, error)
 }
 
 type podmanClient 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) {
@@ -76,10 +80,7 @@ func newPodmanClient(logger *zap.Logger, cfg *Config) (client, error) {
 	c := &podmanClient{
 		conn:     connection,
 		endpoint: fmt.Sprintf("http://d/v%s/libpod", cfg.APIVersion),
-	}
-	err = c.ping()
-	if err != nil {
-		return nil, err
+		timeout:  cfg.Timeout,
 	}
 	return c, nil
 }
@@ -96,11 +97,13 @@ func (c *podmanClient) request(ctx context.Context, path string, params url.Valu
 	return c.conn.Do(req)
 }
 
-func (c *podmanClient) stats() ([]containerStats, error) {
+func (c *podmanClient) stats(ctx context.Context) ([]containerStats, error) {
 	params := url.Values{}
 	params.Add("stream", "false")
 
-	resp, err := c.request(context.Background(), "/containers/stats", params)
+	statsCtx, cancel := context.WithTimeout(ctx, c.timeout)
+	defer cancel()
+	resp, err := c.request(statsCtx, "/containers/stats", params)
 	if err != nil {
 		return nil, err
 	}
@@ -122,8 +125,10 @@ func (c *podmanClient) stats() ([]containerStats, error) {
 	return report.Stats, nil
 }
 
-func (c *podmanClient) ping() error {
-	resp, err := c.request(context.Background(), "/_ping", 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)
 	if err != nil {
 		return err
 	}

receiver/podmanreceiver/podman_client_test.go

@@ -0,0 +1,77 @@
+// Copyright  The 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 podmanreceiver
+
+import (
+	"context"
+	"fmt"
+	"io/ioutil"
+	"net"
+	"os"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"go.uber.org/zap"
+)
+
+func tmpSock(t *testing.T) (net.Listener, string) {
+	f, err := ioutil.TempFile(os.TempDir(), "testsock")
+	if err != nil {
+		t.Fatal(err)
+	}
+	addr := f.Name()
+	os.Remove(addr)
+
+	listener, err := net.Listen("unix", addr)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	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.",
+	)
+}

receiver/podmanreceiver/receiver.go

@@ -72,10 +72,10 @@ func (r *receiver) start(context.Context, component.Host) error {
 	return err
 }
 
-func (r *receiver) scrape(context.Context) (pdata.Metrics, error) {
+func (r *receiver) scrape(ctx context.Context) (pdata.Metrics, error) {
 	var err error
 
-	stats, err := r.client.stats()
+	stats, err := r.client.stats(ctx)
 	if err != nil {
 		r.set.Logger.Error("error fetching stats", zap.Error(err))
 		return pdata.Metrics{}, err

receiver/podmanreceiver/receiver_test.go

@@ -93,14 +93,18 @@ func (c mockClient) factory(logger *zap.Logger, cfg *Config) (client, error) {
 	return c, nil
 }
 
-func (c mockClient) stats() ([]containerStats, error) {
+func (c mockClient) stats(context.Context) ([]containerStats, error) {
 	report := <-c
 	if report.Error != "" {
 		return nil, errors.New(report.Error)
 	}
 	return report.Stats, nil
 }
 
+func (c mockClient) ping(context.Context) error {
+	return nil
+}
+
 type mockConsumer chan pdata.Metrics
 
 func (m mockConsumer) Capabilities() consumer.Capabilities {

receiver/podmanreceiver/testdata/config.yaml

@@ -3,6 +3,7 @@ receivers:
   podman_stats/all:
     endpoint: http://example.com/
     collection_interval: 2s
+    timeout: 20s
 
 processors:
   nop: