internal/components/setup/compose.go - skywalking-infra-e2e - Git at Google

 // Licensed to Apache Software Foundation (ASF) under one or more contributor
 // license agreements. See the NOTICE file distributed with
 // this work for additional information regarding copyright
 // ownership. Apache Software Foundation (ASF) licenses this file to you 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 setup

 import (
 	"context"
 	"errors"
 	"fmt"
 	"os"
 	"regexp"
 	"strconv"
 	"strings"

 	"github.com/docker/go-connections/nat"
 	"github.com/testcontainers/testcontainers-go/wait"

 	"github.com/apache/skywalking-infra-e2e/internal/config"
 	"github.com/apache/skywalking-infra-e2e/internal/logger"
 	"github.com/apache/skywalking-infra-e2e/internal/util"

 	"github.com/docker/docker/api/types"
 	"github.com/docker/docker/api/types/filters"
 	"github.com/docker/docker/client"
 	"github.com/docker/docker/pkg/stdcopy"

 	"github.com/testcontainers/testcontainers-go"
 )

 const (
 	// SeparatorV1 is the separator used in docker-compose v1
 	// refer to https://github.com/docker/compose/blob/5becea4ca9f68875334c92f191a13482bcd6e5cf/compose/service.py#L1492-L1498
 	SeparatorV1 = "_"
 	// SeparatorV2 is the separator used in docker-compose v2
 	// refer to https://github.com/docker/compose/blob/981aea674d052ee1ab252f71c3ca1f9f8a7e32de/pkg/compose/convergence.go#L252-L257
 	SeparatorV2 = "-"
 )

 var (
 	containerNamePattern = regexp.MustCompile(`.*_(?P<containerNum>\d+)$`)
 )

 // ComposeSetup sets up environment according to e2e.yaml.
 func ComposeSetup(e2eConfig *config.E2EConfig) error {
 	composeConfigPath := e2eConfig.Setup.GetFile()
 	if composeConfigPath == "" {
 		return fmt.Errorf("no compose config file was provided")
 	}

 	// build docker client
 	cli, err := client.NewClientWithOpts(client.FromEnv)
 	if err != nil {
 		return err
 	}

 	// setup docker compose
 	composeFilePaths := []string{
 		composeConfigPath,
 	}
 	identifier := GetIdentity()
 	compose := testcontainers.NewLocalDockerCompose(composeFilePaths, identifier)

 	// bind wait port
 	services, err := buildComposeServices(e2eConfig, compose)
 	if err != nil {
 		return fmt.Errorf("bind wait ports error: %v", err)
 	}

 	// build command
 	cmd := make([]string, 0)
 	if e2eConfig.Setup.InitSystemEnvironment != "" {
 		profilePath := util.ResolveAbs(e2eConfig.Setup.InitSystemEnvironment)
 		cmd = append(cmd, "--env-file", profilePath)
 		util.ExportEnvVars(profilePath)
 	}
 	cmd = append(cmd, "up", "-d")

 	// Listen container create
 	listener := NewComposeContainerListener(context.Background(), cli, services)
 	defer listener.Stop()
 	err = listener.Listen(func(container *ComposeContainer) {
 		if err = exposeComposeLog(cli, container.Service, container.ID, logFollower); err == nil {
 			container.Service.beenFollowLog = true
 		}
 	})
 	if err != nil {
 		return err
 	}

 	// setup
 	execError := compose.WithCommand(cmd).Invoke()
 	if execError.Error != nil {
 		return execError.Error
 	}

 	// find exported port and build env
 	err = exposeComposeService(services, cli, identifier, e2eConfig)
 	if err != nil {
 		return err
 	}

 	// run steps
 	err = RunStepsAndWait(e2eConfig.Setup.Steps, e2eConfig.Setup.GetTimeout(), nil)
 	if err != nil {
 		logger.Log.Errorf("execute steps error: %v", err)
 		return err
 	}

 	return nil
 }

 type ComposeService struct {
 	Name           string
 	waitStrategies []*hostPortCachedStrategy
 	beenFollowLog  bool
 }

 func exposeComposeService(services []*ComposeService, cli *client.Client,
 	identity string, e2eConfig *config.E2EConfig) error {
 	dockerProvider := &DockerProvider{client: cli}

 	// find exported port and build env
 	for _, service := range services {
 		// expose port
 		if err := exposeComposePort(dockerProvider, service, cli, identity, e2eConfig); err != nil {
 			return err
 		}

 		// if service log not follow, expose log
 		if !service.beenFollowLog {
 			c, err := service.FindContainer(cli, identity)
 			if err != nil {
 				logger.Log.Warn(err)
 				continue
 			}
 			if err := exposeComposeLog(dockerProvider.client, service, c.ID, logFollower); err != nil {
 				return err
 			}
 			service.beenFollowLog = true
 		}
 	}
 	return nil
 }

 func (c *ComposeService) FindContainer(cli *client.Client, identity string) (*types.Container, error) {
 	serviceName, num := getInstanceName(c.Name)
 	return findContainer(cli, identity, serviceName, num)
 }

 func exposeComposePort(dockerProvider *DockerProvider, service *ComposeService, cli *client.Client, identity string,
 	e2eConfig *config.E2EConfig) error {
 	if len(service.waitStrategies) == 0 {
 		return nil
 	}

 	// get real ip address for access and export to env
 	host, err := dockerProvider.daemonHost(context.Background())
 	if err != nil {
 		return err
 	}

 	container, err := service.FindContainer(cli, identity)
 	if err != nil {
 		return err
 	}

 	// format: <service_name>_host
 	if err := exportComposeEnv(fmt.Sprintf("%s_host", service.Name), host, service.Name); err != nil {
 		return err
 	}

 	for inx := range service.waitStrategies {
 		for _, containerPort := range container.Ports {
 			if int(containerPort.PrivatePort) != service.waitStrategies[inx].expectPort {
 				continue
 			}

 			if err := waitPortUntilReady(e2eConfig, container, dockerProvider, service.waitStrategies[inx].expectPort); err != nil {
 				return err
 			}

 			// expose env config to env
 			// format: <service_name>_<port>
 			if err := exportComposeEnv(
 				fmt.Sprintf("%s_%d", service.Name, containerPort.PrivatePort),
 				fmt.Sprintf("%d", containerPort.PublicPort),
 				service.Name); err != nil {
 				return err
 			}
 			break
 		}
 	}

 	return nil
 }

 // export container log to local path
 func exposeComposeLog(cli *client.Client, service *ComposeService, containerID string, logFollower *util.ResourceLogFollower) error {
 	logs, err := cli.ContainerLogs(logFollower.Ctx, containerID, types.ContainerLogsOptions{
 		ShowStdout: true,
 		ShowStderr: true,
 		Follow:     true,
 		Details:    false,
 	})
 	if err != nil {
 		return err
 	}
 	writer, err := logFollower.BuildLogWriter(fmt.Sprintf("%s/std.log", service.Name))
 	if err != nil {
 		return err
 	}

 	go func() {
 		defer writer.Close()
 		if _, err := stdcopy.StdCopy(writer, writer, logs); err != nil && !errors.Is(err, context.Canceled) {
 			logger.Log.Warnf("write %s std log error: %v", service.Name, err)
 		}
 	}()
 	return nil
 }

 func exportComposeEnv(key, value, service string) error {
 	err := os.Setenv(key, value)
 	if err != nil {
 		return fmt.Errorf("could not set env for %s, %v", service, err)
 	}
 	logger.Log.Infof("export %s=%s", key, value)
 	return nil
 }

 func buildComposeServices(e2eConfig *config.E2EConfig, compose *testcontainers.LocalDockerCompose) ([]*ComposeService, error) {
 	waitTimeout := e2eConfig.Setup.GetTimeout()
 	services := make([]*ComposeService, 0)
 	for service, content := range compose.Services {
 		serviceConfig := content.(map[any]any)
 		ports := serviceConfig["ports"]
 		serviceContext := &ComposeService{Name: service}
 		services = append(services, serviceContext)
 		if ports == nil {
 			continue
 		}

 		portList := ports.([]any)
 		for inx := range portList {
 			exportPort, err := getExpectPort(portList[inx])
 			if err != nil {
 				return nil, err
 			}

 			strategy := &hostPortCachedStrategy{
 				expectPort:       exportPort,
 				HostPortStrategy: *wait.NewHostPortStrategy(nat.Port(fmt.Sprintf("%d/tcp", exportPort))).WithStartupTimeout(waitTimeout),
 			}
 			// temporary don't use testcontainers-go framework wait strategy until fix docker-in-docker bug
 			// compose.WithExposedService(service, exportPort, strategy)
 			serviceContext.waitStrategies = append(serviceContext.waitStrategies, strategy)
 		}
 	}
 	return services, nil
 }

 func getExpectPort(portConfig any) (int, error) {
 	switch conf := portConfig.(type) {
 	case int:
 		return conf, nil
 	case string:
 		portInfo := strings.Split(conf, ":")
 		if len(portInfo) > 1 {
 			return strconv.Atoi(portInfo[1])
 		}
 		return strconv.Atoi(portInfo[0])
 	}
 	return 0, fmt.Errorf("unknown port information: %v", portConfig)
 }

 func findContainer(c *client.Client, projectName, serviceName string, number int) (*types.Container, error) {
 	nameV1 := strings.Join([]string{projectName, serviceName, strconv.Itoa(number)}, SeparatorV1)
 	nameV2 := strings.Join([]string{projectName, serviceName, strconv.Itoa(number)}, SeparatorV2)
 	// filter either names
 	// 1) {project}_{service}_{number}
 	// 2) {project}-{service}-{number}
 	f := filters.NewArgs(filters.Arg("name", nameV1), filters.Arg("name", nameV2))
 	containerListOptions := types.ContainerListOptions{Filters: f}
 	containers, err := c.ContainerList(context.Background(), containerListOptions)
 	if err != nil {
 		return nil, err
 	}

 	if len(containers) == 0 {
 		return nil, fmt.Errorf("could not found container: %s(docker-compose v1) or %s(docker-compose v2)", nameV1, nameV2)
 	}
 	return &containers[0], nil
 }

 func getInstanceName(serviceName string) (service string, number int) {
 	matches := containerNamePattern.FindStringSubmatch(serviceName)
 	if len(matches) == 0 {
 		return serviceName, 1
 	}
 	numberStr := matches[0]
 	number, err := strconv.Atoi(numberStr)
 	if err != nil {
 		return serviceName, 1
 	}
 	return serviceName, number
 }

 // hostPortCachedStrategy cached original target
 type hostPortCachedStrategy struct {
 	wait.HostPortStrategy
 	expectPort int
 	target     wait.StrategyTarget
 }

 func (hp *hostPortCachedStrategy) WaitUntilReady(ctx context.Context, target wait.StrategyTarget) error {
 	hp.target = target
 	return hp.HostPortStrategy.WaitUntilReady(ctx, target)
 }

 func waitPortUntilReady(e2eConfig *config.E2EConfig, container *types.Container, dockerProvider *DockerProvider, expectPort int) error {
 	// wait port
 	waitTimeout := e2eConfig.Setup.GetTimeout()
 	waitPort := nat.Port(fmt.Sprintf("%d/tcp", expectPort))
 	target := &DockerContainer{
 		ID:         container.ID,
 		WaitingFor: wait.NewHostPortStrategy(waitPort),
 		provider:   dockerProvider}
 	return WaitPort(context.Background(), target, waitPort, waitTimeout)
 }
	// Licensed to Apache Software Foundation (ASF) under one or more contributor
	// license agreements. See the NOTICE file distributed with
	// this work for additional information regarding copyright
	// ownership. Apache Software Foundation (ASF) licenses this file to you 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 setup

	import (
	"context"
	"errors"
	"fmt"
	"os"
	"regexp"
	"strconv"
	"strings"

	"github.com/docker/go-connections/nat"
	"github.com/testcontainers/testcontainers-go/wait"

	"github.com/apache/skywalking-infra-e2e/internal/config"
	"github.com/apache/skywalking-infra-e2e/internal/logger"
	"github.com/apache/skywalking-infra-e2e/internal/util"

	"github.com/docker/docker/api/types"
	"github.com/docker/docker/api/types/filters"
	"github.com/docker/docker/client"
	"github.com/docker/docker/pkg/stdcopy"

	"github.com/testcontainers/testcontainers-go"
	)

	const (
	// SeparatorV1 is the separator used in docker-compose v1
	// refer to https://github.com/docker/compose/blob/5becea4ca9f68875334c92f191a13482bcd6e5cf/compose/service.py#L1492-L1498
	SeparatorV1 = "_"
	// SeparatorV2 is the separator used in docker-compose v2
	// refer to https://github.com/docker/compose/blob/981aea674d052ee1ab252f71c3ca1f9f8a7e32de/pkg/compose/convergence.go#L252-L257
	SeparatorV2 = "-"
	)

	var (
	containerNamePattern = regexp.MustCompile(`.*_(?P<containerNum>\d+)$`)
	)

	// ComposeSetup sets up environment according to e2e.yaml.
	func ComposeSetup(e2eConfig *config.E2EConfig) error {
	composeConfigPath := e2eConfig.Setup.GetFile()
	if composeConfigPath == "" {
	return fmt.Errorf("no compose config file was provided")
	}

	// build docker client
	cli, err := client.NewClientWithOpts(client.FromEnv)
	if err != nil {
	return err
	}

	// setup docker compose
	composeFilePaths := []string{
	composeConfigPath,
	}
	identifier := GetIdentity()
	compose := testcontainers.NewLocalDockerCompose(composeFilePaths, identifier)

	// bind wait port
	services, err := buildComposeServices(e2eConfig, compose)
	if err != nil {
	return fmt.Errorf("bind wait ports error: %v", err)
	}

	// build command
	cmd := make([]string, 0)
	if e2eConfig.Setup.InitSystemEnvironment != "" {
	profilePath := util.ResolveAbs(e2eConfig.Setup.InitSystemEnvironment)
	cmd = append(cmd, "--env-file", profilePath)
	util.ExportEnvVars(profilePath)
	}
	cmd = append(cmd, "up", "-d")

	// Listen container create
	listener := NewComposeContainerListener(context.Background(), cli, services)
	defer listener.Stop()
	err = listener.Listen(func(container *ComposeContainer) {
	if err = exposeComposeLog(cli, container.Service, container.ID, logFollower); err == nil {
	container.Service.beenFollowLog = true
	}
	})
	if err != nil {
	return err
	}

	// setup
	execError := compose.WithCommand(cmd).Invoke()
	if execError.Error != nil {
	return execError.Error
	}

	// find exported port and build env
	err = exposeComposeService(services, cli, identifier, e2eConfig)
	if err != nil {
	return err
	}

	// run steps
	err = RunStepsAndWait(e2eConfig.Setup.Steps, e2eConfig.Setup.GetTimeout(), nil)
	if err != nil {
	logger.Log.Errorf("execute steps error: %v", err)
	return err
	}

	return nil
	}

	type ComposeService struct {
	Name string
	waitStrategies []*hostPortCachedStrategy
	beenFollowLog bool
	}

	func exposeComposeService(services []ComposeService, cli client.Client,
	identity string, e2eConfig *config.E2EConfig) error {
	dockerProvider := &DockerProvider{client: cli}

	// find exported port and build env
	for _, service := range services {
	// expose port
	if err := exposeComposePort(dockerProvider, service, cli, identity, e2eConfig); err != nil {
	return err
	}

	// if service log not follow, expose log
	if !service.beenFollowLog {
	c, err := service.FindContainer(cli, identity)
	if err != nil {
	logger.Log.Warn(err)
	continue
	}
	if err := exposeComposeLog(dockerProvider.client, service, c.ID, logFollower); err != nil {
	return err
	}
	service.beenFollowLog = true
	}
	}
	return nil
	}

	func (c ComposeService) FindContainer(cli client.Client, identity string) (*types.Container, error) {
	serviceName, num := getInstanceName(c.Name)
	return findContainer(cli, identity, serviceName, num)
	}

	func exposeComposePort(dockerProvider DockerProvider, service ComposeService, cli *client.Client, identity string,
	e2eConfig *config.E2EConfig) error {
	if len(service.waitStrategies) == 0 {
	return nil
	}

	// get real ip address for access and export to env
	host, err := dockerProvider.daemonHost(context.Background())
	if err != nil {
	return err
	}

	container, err := service.FindContainer(cli, identity)
	if err != nil {
	return err
	}

	// format: <service_name>_host
	if err := exportComposeEnv(fmt.Sprintf("%s_host", service.Name), host, service.Name); err != nil {
	return err
	}

	for inx := range service.waitStrategies {
	for _, containerPort := range container.Ports {
	if int(containerPort.PrivatePort) != service.waitStrategies[inx].expectPort {
	continue
	}

	if err := waitPortUntilReady(e2eConfig, container, dockerProvider, service.waitStrategies[inx].expectPort); err != nil {
	return err
	}

	// expose env config to env
	// format: <service_name>_<port>
	if err := exportComposeEnv(
	fmt.Sprintf("%s_%d", service.Name, containerPort.PrivatePort),
	fmt.Sprintf("%d", containerPort.PublicPort),
	service.Name); err != nil {
	return err
	}
	break
	}
	}

	return nil
	}

	// export container log to local path
	func exposeComposeLog(cli client.Client, service ComposeService, containerID string, logFollower *util.ResourceLogFollower) error {
	logs, err := cli.ContainerLogs(logFollower.Ctx, containerID, types.ContainerLogsOptions{
	ShowStdout: true,
	ShowStderr: true,
	Follow: true,
	Details: false,
	})
	if err != nil {
	return err
	}
	writer, err := logFollower.BuildLogWriter(fmt.Sprintf("%s/std.log", service.Name))
	if err != nil {
	return err
	}

	go func() {
	defer writer.Close()
	if _, err := stdcopy.StdCopy(writer, writer, logs); err != nil && !errors.Is(err, context.Canceled) {
	logger.Log.Warnf("write %s std log error: %v", service.Name, err)
	}
	}()
	return nil
	}

	func exportComposeEnv(key, value, service string) error {
	err := os.Setenv(key, value)
	if err != nil {
	return fmt.Errorf("could not set env for %s, %v", service, err)
	}
	logger.Log.Infof("export %s=%s", key, value)
	return nil
	}

	func buildComposeServices(e2eConfig config.E2EConfig, compose testcontainers.LocalDockerCompose) ([]*ComposeService, error) {
	waitTimeout := e2eConfig.Setup.GetTimeout()
	services := make([]*ComposeService, 0)
	for service, content := range compose.Services {
	serviceConfig := content.(map[any]any)
	ports := serviceConfig["ports"]
	serviceContext := &ComposeService{Name: service}
	services = append(services, serviceContext)
	if ports == nil {
	continue
	}

	portList := ports.([]any)
	for inx := range portList {
	exportPort, err := getExpectPort(portList[inx])
	if err != nil {
	return nil, err
	}

	strategy := &hostPortCachedStrategy{
	expectPort: exportPort,
	HostPortStrategy: *wait.NewHostPortStrategy(nat.Port(fmt.Sprintf("%d/tcp", exportPort))).WithStartupTimeout(waitTimeout),
	}
	// temporary don't use testcontainers-go framework wait strategy until fix docker-in-docker bug
	// compose.WithExposedService(service, exportPort, strategy)
	serviceContext.waitStrategies = append(serviceContext.waitStrategies, strategy)
	}
	}
	return services, nil
	}

	func getExpectPort(portConfig any) (int, error) {
	switch conf := portConfig.(type) {
	case int:
	return conf, nil
	case string:
	portInfo := strings.Split(conf, ":")
	if len(portInfo) > 1 {
	return strconv.Atoi(portInfo[1])
	}
	return strconv.Atoi(portInfo[0])
	}
	return 0, fmt.Errorf("unknown port information: %v", portConfig)
	}

	func findContainer(c client.Client, projectName, serviceName string, number int) (types.Container, error) {
	nameV1 := strings.Join([]string{projectName, serviceName, strconv.Itoa(number)}, SeparatorV1)
	nameV2 := strings.Join([]string{projectName, serviceName, strconv.Itoa(number)}, SeparatorV2)
	// filter either names
	// 1) {project}_{service}_{number}
	// 2) {project}-{service}-{number}
	f := filters.NewArgs(filters.Arg("name", nameV1), filters.Arg("name", nameV2))
	containerListOptions := types.ContainerListOptions{Filters: f}
	containers, err := c.ContainerList(context.Background(), containerListOptions)
	if err != nil {
	return nil, err
	}

	if len(containers) == 0 {
	return nil, fmt.Errorf("could not found container: %s(docker-compose v1) or %s(docker-compose v2)", nameV1, nameV2)
	}
	return &containers[0], nil
	}

	func getInstanceName(serviceName string) (service string, number int) {
	matches := containerNamePattern.FindStringSubmatch(serviceName)
	if len(matches) == 0 {
	return serviceName, 1
	}
	numberStr := matches[0]
	number, err := strconv.Atoi(numberStr)
	if err != nil {
	return serviceName, 1
	}
	return serviceName, number
	}

	// hostPortCachedStrategy cached original target
	type hostPortCachedStrategy struct {
	wait.HostPortStrategy
	expectPort int
	target wait.StrategyTarget
	}

	func (hp *hostPortCachedStrategy) WaitUntilReady(ctx context.Context, target wait.StrategyTarget) error {
	hp.target = target
	return hp.HostPortStrategy.WaitUntilReady(ctx, target)
	}

	func waitPortUntilReady(e2eConfig config.E2EConfig, container types.Container, dockerProvider *DockerProvider, expectPort int) error {
	// wait port
	waitTimeout := e2eConfig.Setup.GetTimeout()
	waitPort := nat.Port(fmt.Sprintf("%d/tcp", expectPort))
	target := &DockerContainer{
	ID: container.ID,
	WaitingFor: wait.NewHostPortStrategy(waitPort),
	provider: dockerProvider}
	return WaitPort(context.Background(), target, waitPort, waitTimeout)
	}