Accept list of docker compose files (#26)

* accept list of docker compose files

* allow tools to be a single string

* allow single string in collectors settings

* ignore missing docker containers

* rename tools log file

Author: lovromazgon

.gitignore

@@ -24,3 +24,4 @@
 
 ### Benchi ###
 /results
+/benchi

README.md

@@ -69,7 +69,7 @@ and tool combination). Each benchmark run folder will contain:
 
 - `infra_NAME.log`: Log file containing the output of the infrastructure docker
   containers, split per infra service.
-- `tools.log`: Log file containing the output of the tools docker containers.
+- `tool_NAME.log`: Log file containing the output of the tool docker containers.
 - `COLLECTOR.csv`: Raw metrics collected using the corresponding
   [metrics collector](#collectors).
 

config/config.go

@@ -33,7 +33,7 @@ type ServiceConfig struct {
 	// DockerCompose is the path to the Docker Compose file for the service. If
 	// it's a relative path, it will be resolved relative to the configuration
 	// file.
-	DockerCompose string `yaml:"compose"`
+	DockerCompose StringList `yaml:"compose"`
 }
 
 // Infrastructure represents a map of service configurations for the infrastructure.
@@ -51,7 +51,7 @@ type MetricsCollector struct {
 	Settings map[string]any `yaml:"settings"`
 	// Tools is a list of tools for which the collector is applicable. If empty,\
 	// the collector will be applied to all tools.
-	Tools []string `yaml:"tools"`
+	Tools StringList `yaml:"tools"`
 }
 
 // Test represents the configuration for a test.
@@ -129,5 +129,5 @@ type TestHook struct {
 	Run string `yaml:"run"`
 	// Tools is a list of tools for which the hook is applicable. If empty, the
 	// hook will be applied to all tools.
-	Tools []string `yaml:"tools"`
+	Tools StringList `yaml:"tools"`
 }

config/string_list.go

@@ -0,0 +1,44 @@
+// Copyright © 2025 Meroxa, Inc.
+//
+// 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 config
+
+import (
+	"fmt"
+
+	"gopkg.in/yaml.v3"
+)
+
+// StringList represents a YAML field that can be either a string or a slice of strings.
+type StringList []string
+
+// UnmarshalYAML implements the yaml.Unmarshaler interface.
+func (s *StringList) UnmarshalYAML(value *yaml.Node) error {
+	var single string
+	var multi []string
+
+	// Try to unmarshal as a single string
+	if err := value.Decode(&single); err == nil {
+		*s = []string{single}
+		return nil
+	}
+
+	// Try to unmarshal as a slice of strings
+	if err := value.Decode(&multi); err == nil {
+		*s = multi
+		return nil
+	}
+
+	return fmt.Errorf("failed to unmarshal StringList")
+}

config/string_list_test.go

@@ -0,0 +1,53 @@
+// Copyright © 2025 Meroxa, Inc.
+//
+// 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 config
+
+import (
+	"reflect"
+	"testing"
+
+	"gopkg.in/yaml.v3"
+)
+
+func TestStringOrSlice_UnmarshalYAML(t *testing.T) {
+	testCases := []struct {
+		input string
+		want  StringList
+	}{{
+		input: `"single"`,
+		want:  StringList{"single"},
+	}, {
+		input: `["one", "two", "three"]`,
+		want:  StringList{"one", "two", "three"},
+	}, {
+		input: `- foo
+- bar
+- baz`,
+		want: StringList{"foo", "bar", "baz"},
+	}}
+
+	for _, tc := range testCases {
+		t.Run(tc.input, func(t *testing.T) {
+			var result StringList
+			err := yaml.Unmarshal([]byte(tc.input), &result)
+			if err != nil {
+				t.Fatalf("unexpected error: %v", err)
+			}
+			if !reflect.DeepEqual(result, tc.want) {
+				t.Errorf("got: %v, want: %v", result, tc.want)
+			}
+		})
+	}
+}

example/README.md

@@ -38,7 +38,7 @@ The output folder will contain logs and results:
   `kafka-to-kafka` test and the `conduit` tool.
   - `infra_kafka.log`: Log file containing the output of the kafka infrastructure
     docker containers.
-  - `tools.log`: Log file containing the output of the tools docker containers
+  - `tool_conduit.log`: Log file containing the output of the tool docker containers
     (conduit).
   - `conduit.csv`: Metrics collected using the [Conduit](../README.md#conduit) collector.
   - `docker.csv`: Metrics collected using the [Docker](../README.md##docker) collector.

metrics/conduit/config.go

@@ -17,12 +17,13 @@ package conduit
 import (
 	"fmt"
 
+	"github.com/conduitio/benchi/config"
 	"github.com/go-viper/mapstructure/v2"
 )
 
 type Config struct {
 	// Pipelines is a list of pipelines to monitor.
-	Pipelines []string `yaml:"pipelines"`
+	Pipelines config.StringList `yaml:"pipelines"`
 }
 
 func parseConfig(settings map[string]any) (Config, error) {

metrics/docker/collector.go

@@ -87,14 +87,15 @@ func (c *Collector) Configure(settings map[string]any) error {
 }
 
 func (c *Collector) Run(ctx context.Context) error {
-	out := make(chan stats, 100+len(c.cfg.Containers))
+	out := make(chan stats, 4*len(c.cfg.Containers))
 	var wg sync.WaitGroup
 
 	for _, container := range c.cfg.Containers {
 		wg.Add(1)
 		go func() {
 			defer wg.Done()
-			collect(ctx, c.logger, c.dockerClient, container, out)
+			logger := c.logger.With("container", container)
+			collect(ctx, logger, c.dockerClient, container, out)
 		}()
 	}
 
@@ -110,6 +111,7 @@ func (c *Collector) Run(ctx context.Context) error {
 			if firstError == nil {
 				firstError = s.Err
 			}
+			continue
 		}
 
 		c.storeStatsEntry(s.statsEntry)

metrics/docker/config.go

@@ -17,12 +17,13 @@ package docker
 import (
 	"fmt"
 
+	"github.com/conduitio/benchi/config"
 	"github.com/go-viper/mapstructure/v2"
 )
 
 type Config struct {
 	// Containers is a list of containers to monitor.
-	Containers []string `yaml:"containers"`
+	Containers config.StringList `yaml:"containers"`
 }
 
 func parseConfig(settings map[string]any) (Config, error) {

metrics/docker/stats_helpers.go

@@ -28,6 +28,7 @@ import (
 
 	"github.com/docker/docker/api/types/container"
 	"github.com/docker/docker/client"
+	"github.com/docker/docker/errdefs"
 )
 
 // statsEntry represents the statistics data collected from a container.
@@ -63,10 +64,14 @@ func collect(
 	containerName string,
 	out chan<- stats,
 ) {
-	logger.Debug("Collecting stats", "container", containerName)
+	logger.Debug("Collecting stats")
 
 	response, err := cli.ContainerStats(ctx, containerName, true)
 	if err != nil {
+		if errdefs.IsNotFound(err) {
+			logger.Warn("Container not found, metrics won't be collected for container", "error", err)
+			return
+		}
 		out <- stats{Container: containerName, Err: err}
 		return
 	}
@@ -94,7 +99,7 @@ func collect(
 			if err == io.EOF {
 				break
 			}
-			out <- stats{Err: err}
+			out <- stats{Container: containerName, Err: err}
 			time.Sleep(100 * time.Millisecond)
 			continue
 		}

metrics/kafka/config.go

@@ -17,12 +17,13 @@ package kafka
 import (
 	"fmt"
 
+	"github.com/conduitio/benchi/config"
 	"github.com/go-viper/mapstructure/v2"
 )
 
 type Config struct {
 	// Topics is a list of topics to monitor.
-	Topics []string `yaml:"topics"`
+	Topics config.StringList `yaml:"topics"`
 }
 
 func parseConfig(settings map[string]any) (Config, error) {

runner.go

@@ -603,7 +603,7 @@ func (r *TestRunner) runTool(ctx context.Context) (err error) {
 		return nil
 	}
 
-	logPath := filepath.Join(r.resultsDir, "tools.log")
+	logPath := filepath.Join(r.resultsDir, fmt.Sprintf("tool_%s.log", r.tool))
 
 	err = r.dockerComposeUpWait(ctx, logger, paths, r.toolContainers, logPath)
 	if err != nil {
@@ -1011,9 +1011,9 @@ func (r *TestRunner) pullHookImages(ctx context.Context, logger *slog.Logger, ho
 }
 
 func collectDockerComposeFiles(cfgs ...config.ServiceConfig) []string {
-	paths := make([]string, len(cfgs))
-	for i, cfg := range cfgs {
-		paths[i] = cfg.DockerCompose
+	paths := make([]string, 0, len(cfgs))
+	for _, cfg := range cfgs {
+		paths = append(paths, cfg.DockerCompose...)
 	}
 	return paths
 }