Add support for multiple Docker hosts

CONTRIBUTING.md

@@ -120,3 +120,10 @@ docker run --rm -it \
 export CUPDATE_OTEL_TARGET=localhost:4317
 export CUPDATE_OTEL_INSECURE=true
 ```
+
+Optionally proxy a Docker socket to test Docker over TCP. Use the proxied port
+as the Docker host rather then the one specified in `.env-docker`.
+
+```shell
+go run tools/sockproxy/*.go -p 3000 docker.sock
+```

README.md

@@ -24,6 +24,7 @@ Docker in the [cookbook](docs/cookbook/README.md).
 
 Features:
 
+- Supports Kubernetes and Docker (one or more hosts, local or remote)
 - Zero configuration required
 - Performant and lightweight - uses virtually zero CPU and roughly 14MiB RAM
 - Auto-detect container images in Kubernetes and Docker

cmd/cupdate/main.go

@@ -78,8 +78,8 @@ type Config struct {
 	} `envPrefix:"KUBERNETES_"`
 
 	Docker struct {
-		Host                 string `env:"HOST"`
-		IncludeAllContainers bool   `env:"INCLUDE_ALL_CONTAINERS"`
+		Hosts                []string `env:"HOST"`
+		IncludeAllContainers bool     `env:"INCLUDE_ALL_CONTAINERS"`
 	} `envPrefix:"DOCKER_"`
 
 	OTEL struct {
@@ -135,7 +135,7 @@ func main() {
 	// Set up the configured platform (Docker if specified, auto discovery of
 	// Kubernetes otherwise)
 	var targetPlatform platform.Grapher
-	if config.Docker.Host == "" {
+	if len(config.Docker.Hosts) == 0 {
 		var kubernetesConfig *rest.Config
 		if config.Kubernetes.Host == "" {
 			var err error
@@ -156,12 +156,20 @@ func main() {
 			os.Exit(1)
 		}
 	} else {
-		targetPlatform, err = docker.NewPlatform(context.Background(), config.Docker.Host, &docker.Options{
-			IncludeAllContainers: config.Docker.IncludeAllContainers,
-		})
-		if err != nil {
-			slog.ErrorContext(ctx, "Failed to create docker source", slog.Any("error", err))
-			os.Exit(1)
+		graphers := make([]platform.Grapher, 0)
+		for _, host := range config.Docker.Hosts {
+			platform, err := docker.NewPlatform(context.Background(), host, &docker.Options{
+				IncludeAllContainers: config.Docker.IncludeAllContainers,
+			})
+			if err != nil {
+				slog.ErrorContext(ctx, "Failed to create docker source", slog.Any("error", err))
+				os.Exit(1)
+			}
+
+			graphers = append(graphers, platform)
+		}
+		targetPlatform = &platform.CompoundGrapher{
+			Graphers: graphers,
 		}
 	}
 

docs/config.md

@@ -25,7 +25,7 @@ done using environment variables.
 | `CUPDATE_PROCESSING_QUEUE_RATE`           | The desired processing rate under normal circumstances.                                                                                                 | `1m`                            |
 | `CUPDATE_KUBERNETES_HOST`                 | The host of the Kubernetes API. For use with proxying.                                                                                                  | Required to use Kubernetes.     |
 | `CUPDATE_KUBERNETES_INCLUDE_OLD_REPLICAS` | Whether or not to include old replica sets when scraping.                                                                                               | `false`                         |
-| `CUPDATE_DOCKER_HOST`                     | Docker host address.                                                                                                                                    | Required to use Docker.         |
+| `CUPDATE_DOCKER_HOST`                     | One or more comma-separated Docker host URIs. Supports unix://path and tcp://host:port URIs.                                                            | Required to use Docker.         |
 | `CUPDATE_DOCKER_INCLUDE_ALL_CONTAINERS`   | Whether or not to include containers in any state, not just running containers.                                                                         | `false`                         |
 | `CUPDATE_OTEL_TARGET`                     | Target URL to an Open Telemetry GRPC ingest endpoint.                                                                                                   | Required to use Open Telemetry. |
 | `CUPDATE_OTEL_INSECURE`                   | Disable client transport security for the Open Telemetry GRPC connection.                                                                               | `false`                         |

go.mod

@@ -61,6 +61,7 @@ require (
 	github.com/prometheus/common v0.55.0 // indirect
 	github.com/prometheus/procfs v0.15.1 // indirect
 	github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec // indirect
+	github.com/stretchr/objx v0.5.2 // indirect
 	github.com/x448/float16 v0.8.4 // indirect
 	go.opentelemetry.io/auto/sdk v1.1.0 // indirect
 	go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.33.0 // indirect

go.sum

@@ -106,6 +106,8 @@ github.com/rogpeppe/go-internal v1.13.1/go.mod h1:uMEvuHeurkdAXX61udpOXGD/AzZDWN
 github.com/spf13/pflag v1.0.5 h1:iy+VFUOCP1a+8yFto/drg2CJ5u0yRoB7fZw3DKv/JXA=
 github.com/spf13/pflag v1.0.5/go.mod h1:McXfInJRrz4CZXVZOBLb0bTZqETkiAhM9Iw0y3An2Bg=
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
+github.com/stretchr/objx v0.5.2 h1:xuMeJ0Sdp5ZMRXx/aWO6RZxdr3beISkG5/G/aIRr3pY=
+github.com/stretchr/objx v0.5.2/go.mod h1:FRsXN1f5AsAjCGJKqEizvkpNtU+EGNCLh3NxZ/8L+MA=
 github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
 github.com/stretchr/testify v1.10.0 h1:Xv5erBjTwe/5IxqUQTdXv5kgmIvbHo3QQyRwhJsOfJA=
 github.com/stretchr/testify v1.10.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY=

internal/platform/docker/platform.go

@@ -8,6 +8,7 @@ import (
 	"net"
 	"net/http"
 	"net/url"
+	"os"
 	"strings"
 	"time"
 
@@ -28,25 +29,47 @@ type Options struct {
 	IncludeAllContainers bool
 }
 
-func NewPlatform(ctx context.Context, host string, options *Options) (*Platform, error) {
+func NewPlatform(ctx context.Context, dockerURI string, options *Options) (*Platform, error) {
 	if options == nil {
 		options = &Options{}
 	}
 
-	if !strings.HasPrefix(host, "unix://") {
-		return nil, fmt.Errorf("unexpected docker host - expected a unix socket")
-	}
-	path := strings.TrimPrefix(host, "unix://")
+	var transport *http.Transport
+	if strings.HasPrefix(dockerURI, "unix://") {
+		host := strings.TrimPrefix(dockerURI, "unix://")
 
-	client := &http.Client{
-		Transport: &http.Transport{
+		if _, err := os.Stat(host); err != nil {
+			return nil, err
+		}
+
+		transport = &http.Transport{
+			DialContext: func(ctx context.Context, _, _ string) (net.Conn, error) {
+				return (&net.Dialer{
+					Timeout: 5 * time.Second,
+				}).DialContext(ctx, "unix", host)
+			},
+		}
+	} else if strings.HasPrefix(dockerURI, "tcp://") {
+		host := strings.TrimPrefix(dockerURI, "tcp://")
+
+		if _, _, err := net.SplitHostPort(host); err != nil {
+			return nil, err
+		}
+
+		transport = &http.Transport{
 			DialContext: func(ctx context.Context, _, _ string) (net.Conn, error) {
 				return (&net.Dialer{
 					Timeout: 5 * time.Second,
-				}).DialContext(ctx, "unix", path)
+				}).DialContext(ctx, "tcp", host)
 			},
-		},
-		Timeout: 10 * time.Second,
+		}
+	} else {
+		return nil, fmt.Errorf("unsupported docker URI: %s", dockerURI)
+	}
+
+	client := &http.Client{
+		Transport: transport,
+		Timeout:   10 * time.Second,
 	}
 
 	p := &Platform{
@@ -55,7 +78,7 @@ func NewPlatform(ctx context.Context, host string, options *Options) (*Platform,
 		includeAllContainers: options.IncludeAllContainers,
 	}
 
-	// Make sure that we can connect to the socket.
+	// Make sure that we can connect to the host.
 	// For now, we probably support most API versions - no need to limit the use
 	// or pin to specific API versions using docker's versioned path prefix
 	_, _, err := p.GetVersion(ctx)
@@ -67,7 +90,7 @@ func NewPlatform(ctx context.Context, host string, options *Options) (*Platform,
 }
 
 func (p *Platform) GetVersion(ctx context.Context) (string, string, error) {
-	req, err := http.NewRequestWithContext(ctx, http.MethodGet, "http://unix/version", nil)
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, "http://_/version", nil)
 	if err != nil {
 		return "", "", err
 	}
@@ -112,7 +135,7 @@ func (p *Platform) GetContainers(ctx context.Context, options *GetContainersOpti
 		query.Set("filters", string(filters))
 	}
 
-	req, err := http.NewRequestWithContext(ctx, http.MethodGet, "http://unix/containers/json?"+query.Encode(), nil)
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, "http://_/containers/json?"+query.Encode(), nil)
 	if err != nil {
 		return nil, err
 	}
@@ -136,7 +159,7 @@ func (p *Platform) GetContainers(ctx context.Context, options *GetContainersOpti
 }
 
 func (p *Platform) GetImage(ctx context.Context, nameOrID string) (*Image, error) {
-	req, err := http.NewRequestWithContext(ctx, http.MethodGet, "http://unix/images/"+url.PathEscape(nameOrID)+"/json", nil)
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, "http://_/images/"+url.PathEscape(nameOrID)+"/json", nil)
 	if err != nil {
 		return nil, err
 	}

internal/platform/platform.go

@@ -2,8 +2,10 @@ package platform
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"log/slog"
+	"sync"
 	"time"
 
 	"github.com/AlexGustafsson/cupdate/internal/graph"
@@ -89,6 +91,39 @@ func (g *PollGrapher) GraphContinously(ctx context.Context) (<-chan Graph, error
 	return ch, nil
 }
 
+// CompoundGrapher creates a graph from one or more [Grapher] simultaneously.
+type CompoundGrapher struct {
+	Graphers []Grapher
+}
+
+func (g *CompoundGrapher) Graph(ctx context.Context) (Graph, error) {
+	graphs := make([]Graph, len(g.Graphers))
+	errs := make([]error, len(g.Graphers))
+
+	// Don't use ErrGroup in order to retain all errors to help users debug any
+	// issues
+	var wg sync.WaitGroup
+	for i, grapher := range g.Graphers {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			graphs[i], errs[i] = grapher.Graph(ctx)
+		}()
+	}
+	wg.Wait()
+
+	if err := errors.Join(errs...); err != nil {
+		return nil, err
+	}
+
+	compoundGraph := NewGraph()
+	for _, graph := range graphs {
+		compoundGraph.InsertGraph(graph)
+	}
+
+	return compoundGraph, nil
+}
+
 func NewGraph() Graph {
 	return graph.New[Node]()
 }

internal/platform/platform_test.go

@@ -0,0 +1,128 @@
+package platform
+
+import (
+	"context"
+	"errors"
+	"strings"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/mock"
+	"github.com/stretchr/testify/require"
+)
+
+var _ Grapher = (*MockGrapher)(nil)
+
+type MockGrapher struct {
+	mock.Mock
+}
+
+// Graph implements Grapher.
+func (m *MockGrapher) Graph(ctx context.Context) (Graph, error) {
+	args := m.Called(ctx)
+	return args.Get(0).(Graph), args.Error(1)
+}
+
+var _ Node = (*TestNode)(nil)
+
+type TestNode struct {
+	id       string
+	nodeType string
+}
+
+// ID implements Node.
+func (m *TestNode) ID() string {
+	return m.id
+}
+
+// Type implements Node.
+func (m *TestNode) Type() string {
+	return m.nodeType
+}
+
+func TestCompoundGrapherHappyPath(t *testing.T) {
+	graph1 := NewGraph()
+	graph1.InsertTree(
+		&TestNode{id: "graph1/node1", nodeType: "testnode"},
+		&TestNode{id: "graph1/node2", nodeType: "testnode"},
+	)
+
+	grapher1 := &MockGrapher{}
+	grapher1.On("Graph", mock.Anything).Return(graph1, nil)
+
+	graph2 := NewGraph()
+	graph2.InsertTree(
+		&TestNode{id: "graph2/node1", nodeType: "testnode"},
+		&TestNode{id: "graph2/node2", nodeType: "testnode"},
+	)
+
+	grapher2 := &MockGrapher{}
+	grapher2.On("Graph", mock.Anything).Return(graph2, nil)
+
+	compoundGrapher := CompoundGrapher{
+		Graphers: []Grapher{grapher1, grapher2},
+	}
+
+	graph, err := compoundGrapher.Graph(context.TODO())
+	require.NoError(t, err)
+	grapher1.AssertExpectations(t)
+	grapher2.AssertExpectations(t)
+
+	expectedString := `graph1/node1->graph1/node2
+graph2/node1->graph2/node2`
+
+	actualString := graph.String()
+
+	// Ignore order when matching
+	expected := strings.Split(expectedString, "\n")
+	actual := strings.Split(actualString, "\n")
+
+	assert.ElementsMatch(t, expected, actual)
+}
+
+func TestCompoundGrapherError(t *testing.T) {
+	testCases := []struct {
+		Name        string
+		Err1        error
+		Err2        error
+		ExpectedErr string
+	}{
+		{
+			Name:        "grapher 1 fails",
+			Err1:        errors.New("failed to graph"),
+			Err2:        nil,
+			ExpectedErr: "failed to graph",
+		},
+		{
+			Name:        "grapher 2 fails",
+			Err1:        nil,
+			Err2:        errors.New("failed to graph"),
+			ExpectedErr: "failed to graph",
+		},
+		{
+			Name:        "grapher 1 and 2 fail",
+			Err1:        errors.New("failed to graph"),
+			Err2:        errors.New("failed to graph"),
+			ExpectedErr: "failed to graph\nfailed to graph",
+		},
+	}
+
+	for _, testCase := range testCases {
+		t.Run(testCase.Name, func(t *testing.T) {
+			grapher1 := &MockGrapher{}
+			grapher1.On("Graph", mock.Anything).Return(Graph(nil), testCase.Err1)
+
+			grapher2 := &MockGrapher{}
+			grapher2.On("Graph", mock.Anything).Return(Graph(nil), testCase.Err2)
+
+			compoundGrapher := CompoundGrapher{
+				Graphers: []Grapher{grapher1, grapher2},
+			}
+
+			_, err := compoundGrapher.Graph(context.TODO())
+			require.EqualError(t, err, testCase.ExpectedErr)
+			grapher1.AssertExpectations(t)
+			grapher2.AssertExpectations(t)
+		})
+	}
+}