Merge branch 'dev' into thean/s3-ipv6-client-flag

Author: TheanLim

agent/config/config.go

@@ -240,7 +240,8 @@ func NewConfig(ec2client ec2.EC2MetadataClient) (*Config, error) {
 		ec2client = ec2.NewBlackholeEC2MetadataClient()
 	}
 
-	config.determineIPCompatibility(ec2client)
+	// TODO feat:IPv6-only - Enable when launching IPv6-only support
+	// config.determineIPCompatibility(ec2client)
 
 	if config.complete() {
 		// No need to do file / network IO

agent/config/config_unix.go

@@ -170,6 +170,10 @@ func getConfigFileName() (string, error) {
 // determining the IP compatibility of the container instance.
 // This is a fallback to help with graceful adoption of Agent in IPv6-only environments
 // without disrupting existing environments.
+//
+// TODO feat:IPv6-only - Remove lint rule below
+//
+//lint:ignore U1000 Function will be used in the future
 func (c *Config) determineIPCompatibility(ec2client ec2.EC2MetadataClient) {
 	// Load primary ENI's MAC address on EC2 Launch Type
 	var primaryENIMAC string

agent/config/config_unix_test.go

@@ -384,6 +384,8 @@ func TestDetermineIPCompatibility(t *testing.T) {
 // Tests that IPCompatibility defaults to IPv4-only when determining IP compatibility of
 // the container instance fails due to some error.
 func TestIPCompatibilityFallback(t *testing.T) {
+	// TODO feat:IPv6-only - Remove skip
+	t.Skip("Enable when launching IPv6-only support")
 	defer setTestRegion()()
 	ctrl := gomock.NewController(t)
 	mockEc2Metadata := mock_ec2.NewMockEC2MetadataClient(ctrl)

agent/config/ipcompatibility/ipcompatibility.go

@@ -24,11 +24,16 @@ func NewIPCompatibility(ipv4Compatible, ipv6Compatible bool) IPCompatibility {
 	return IPCompatibility{ipv4Compatible: ipv4Compatible, ipv6Compatible: ipv6Compatible}
 }
 
-// Returns an IPv4-only IPCompatibility instance.
+// Returns an IPv4-only IPCompatibility value.
 func NewIPv4OnlyCompatibility() IPCompatibility {
 	return NewIPCompatibility(true, false)
 }
 
+// Returns an IPv6-only IPCompatibility value.
+func NewIPv6OnlyCompatibility() IPCompatibility {
+	return NewIPCompatibility(false, true)
+}
+
 // IsIPv4Compatible returns the current IPv4 compatibility status.
 func (ic *IPCompatibility) IsIPv4Compatible() bool {
 	return ic.ipv4Compatible

agent/config/ipcompatibility/ipcompatibility_test.go

@@ -44,6 +44,12 @@ func TestIPv4OnlyCompatibility(t *testing.T) {
 	assert.False(t, c.IsIPv6Compatible())
 }
 
+func TestIPv6OnlyCompatibility(t *testing.T) {
+	c := NewIPv6OnlyCompatibility()
+	assert.True(t, c.IsIPv6Compatible())
+	assert.False(t, c.IsIPv4Compatible())
+}
+
 func TestIsIPv6Only(t *testing.T) {
 	tests := []struct {
 		name     string

agent/dockerclient/dockerapi/docker_client.go

@@ -1300,6 +1300,7 @@ func (dg *dockerGoClient) Version(ctx context.Context, timeout time.Duration) (s
 	}
 
 	version = info.Version
+	seelog.Debugf("Determined the Docker server version: %s", version)
 	dg.setDaemonVersion(version)
 	return version, nil
 }

agent/engine/docker_task_engine.go

@@ -56,8 +56,10 @@ import (
 	"github.com/aws/amazon-ecs-agent/ecs-agent/logger/field"
 	"github.com/aws/amazon-ecs-agent/ecs-agent/utils/retry"
 	"github.com/aws/amazon-ecs-agent/ecs-agent/utils/ttime"
+	"github.com/aws/aws-sdk-go-v2/service/cloudwatchlogs"
 	"github.com/aws/aws-sdk-go/aws"
 	ep "github.com/aws/aws-sdk-go/aws/endpoints"
+	"github.com/aws/smithy-go/ptr"
 	"github.com/docker/docker/api/types"
 	dockercontainer "github.com/docker/docker/api/types/container"
 	"github.com/docker/docker/api/types/registry"
@@ -89,6 +91,8 @@ const (
 	logDriverTypeFirelens = "awsfirelens"
 	logDriverTypeFluentd  = "fluentd"
 	logDriverTypeAwslogs  = "awslogs"
+	awsLogsEndpointKey    = "awslogs-endpoint"
+	awsLogsRegionKey      = "awslogs-region"
 	logDriverTag          = "tag"
 	logDriverMode         = "mode"
 	logDriverBufferSize   = "max-buffer-size"
@@ -97,6 +101,7 @@ const (
 	logDriverFluentdAddress     = "fluentd-address"
 	dataLogDriverPath           = "/data/firelens/"
 	logDriverAsyncConnect       = "fluentd-async-connect"
+	logDriverAsync              = "fluentd-async"
 	logDriverSubSecondPrecision = "fluentd-sub-second-precision"
 	logDriverBufferLimit        = "fluentd-buffer-limit"
 	dataLogDriverSocketPath     = "/socket/fluent.sock"
@@ -1874,7 +1879,35 @@ func (engine *DockerTaskEngine) createContainer(task *apitask.Task, container *a
 	// Update the environment variables FLUENT_HOST and FLUENT_PORT depending on the supported network modes - bridge
 	// and awsvpc. For reference - https://docs.docker.com/config/containers/logging/fluentd/.
 	if hostConfig.LogConfig.Type == logDriverTypeFirelens {
-		hostConfig.LogConfig = getFirelensLogConfig(task, container, hostConfig, engine.cfg)
+		// We need the Docker server version in order to generate the appropriate firelens log config
+		dockerServerVersion, err := engine.Version()
+		if err != nil {
+			logger.Error("Failed to determine Docker server version for Firelens log config generation", logger.Fields{
+				field.TaskID:    task.GetID(),
+				field.Container: container.Name,
+				field.Error:     err,
+			})
+			return dockerapi.DockerContainerMetadata{
+				Error: dockerapi.CannotCreateContainerError{FromError: errors.Wrapf(versionErr,
+					"failed to create container - container uses awsfirelens log driver and we failed to "+
+						"determine the Docker server version for Firelens log config generation")},
+			}
+		}
+		logConfig, err := getFirelensLogConfig(task, container, hostConfig, engine.cfg, dockerServerVersion)
+		if err != nil {
+			logger.Error("Failed to generate the Firelens log config", logger.Fields{
+				field.TaskID:    task.GetID(),
+				field.Container: container.Name,
+				field.Error:     err,
+			})
+			return dockerapi.DockerContainerMetadata{
+				Error: dockerapi.CannotCreateContainerError{FromError: errors.Wrapf(err,
+					"failed to create container - container uses awsfirelens log driver and we failed to "+
+						"generate the Firelens log config")},
+			}
+		}
+		hostConfig.LogConfig = logConfig
+
 		if task.IsNetworkModeAWSVPC() {
 			container.MergeEnvironmentVariables(map[string]string{
 				fluentNetworkHost: FluentAWSVPCHostValue,
@@ -1916,29 +1949,33 @@ func (engine *DockerTaskEngine) createContainer(task *apitask.Task, container *a
 
 	// This is a short term solution only for specific regions
 	if hostConfig.LogConfig.Type == logDriverTypeAwslogs {
-		region := engine.cfg.AWSRegion
-		if _, ok := unresolvedIsolatedRegions[region]; ok {
-			endpoint := ""
-			dnsSuffix := ""
-			partition, ok := ep.PartitionForRegion(ep.DefaultPartitions(), region)
-			if !ok {
-				logger.Warn("No partition resolved for region. Using AWS default", logger.Fields{
-					"region":           region,
-					"defaultDNSSuffix": ep.AwsPartition().DNSSuffix(),
-				})
-				dnsSuffix = ep.AwsPartition().DNSSuffix()
-			} else {
-				resolvedEndpoint, err := partition.EndpointFor("logs", region)
-				if err == nil {
-					endpoint = resolvedEndpoint.URL
+		if engine.cfg.InstanceIPCompatibility.IsIPv6Only() {
+			engine.setAWSLogsDualStackEndpoint(task, container, hostConfig)
+		} else {
+			region := engine.cfg.AWSRegion
+			if _, ok := unresolvedIsolatedRegions[region]; ok {
+				endpoint := ""
+				dnsSuffix := ""
+				partition, ok := ep.PartitionForRegion(ep.DefaultPartitions(), region)
+				if !ok {
+					logger.Warn("No partition resolved for region. Using AWS default", logger.Fields{
+						"region":           region,
+						"defaultDNSSuffix": ep.AwsPartition().DNSSuffix(),
+					})
+					dnsSuffix = ep.AwsPartition().DNSSuffix()
 				} else {
-					dnsSuffix = partition.DNSSuffix()
+					resolvedEndpoint, err := partition.EndpointFor("logs", region)
+					if err == nil {
+						endpoint = resolvedEndpoint.URL
+					} else {
+						dnsSuffix = partition.DNSSuffix()
+					}
 				}
+				if endpoint == "" {
+					endpoint = fmt.Sprintf("https://logs.%s.%s", region, dnsSuffix)
+				}
+				hostConfig.LogConfig.Config[awsLogsEndpointKey] = endpoint
 			}
-			if endpoint == "" {
-				endpoint = fmt.Sprintf("https://logs.%s.%s", region, dnsSuffix)
-			}
-			hostConfig.LogConfig.Config["awslogs-endpoint"] = endpoint
 		}
 	}
 
@@ -2108,29 +2145,64 @@ func (engine *DockerTaskEngine) createContainer(task *apitask.Task, container *a
 	return metadata
 }
 
-func getFirelensLogConfig(task *apitask.Task, container *apicontainer.Container,
-	hostConfig *dockercontainer.HostConfig, cfg *config.Config) dockercontainer.LogConfig {
+// getFirelensLogConfig generates the fluentd log driver's log config.
+// Every container that wants to use Firelens for logging, gets one instance of the fluentd log driver associated to it.
+func getFirelensLogConfig(task *apitask.Task,
+	container *apicontainer.Container,
+	hostConfig *dockercontainer.HostConfig,
+	cfg *config.Config, dockerServerVersion string) (dockercontainer.LogConfig, error) {
+	var firelensConfig dockercontainer.LogConfig
+	// Set the log driver type
+	firelensConfig.Type = logDriverTypeFluentd
+	// Start from the existing container host config
+	hostConfigLogConfig := hostConfig.LogConfig
+	// Initialize a config to store the different log driver options
+	firelensConfig.Config = make(map[string]string)
+	// Generate a tag based on the task ID
 	fields := strings.Split(task.Arn, "/")
 	taskID := fields[len(fields)-1]
 	tag := fmt.Sprintf(fluentTagDockerFormat, container.Name, taskID)
+	firelensConfig.Config[logDriverTag] = tag
+	// Construct the fluent socket address
 	fluentd := socketPathPrefix + filepath.Join(cfg.DataDirOnHost, dataLogDriverPath, taskID, dataLogDriverSocketPath)
-	logConfig := hostConfig.LogConfig
-	bufferLimit, bufferLimitExists := logConfig.Config[apitask.FirelensLogDriverBufferLimitOption]
-	logConfig.Type = logDriverTypeFluentd
-	logConfig.Config = make(map[string]string)
-	logConfig.Config[logDriverTag] = tag
-	logConfig.Config[logDriverFluentdAddress] = fluentd
-	logConfig.Config[logDriverAsyncConnect] = strconv.FormatBool(cfg.FirelensAsyncEnabled.Enabled())
-	logConfig.Config[logDriverSubSecondPrecision] = strconv.FormatBool(true)
+	firelensConfig.Config[logDriverFluentdAddress] = fluentd
+	// Enable sub-second precision
+	firelensConfig.Config[logDriverSubSecondPrecision] = strconv.FormatBool(true)
+	// Set the log driver buffer limit if passed via the task payload
+	bufferLimit, bufferLimitExists := hostConfigLogConfig.Config[apitask.FirelensLogDriverBufferLimitOption]
 	if bufferLimitExists {
-		logConfig.Config[logDriverBufferLimit] = bufferLimit
+		firelensConfig.Config[logDriverBufferLimit] = bufferLimit
+	}
+	// Determine whether to use the "fluentd-async" option or the legacy "fluentd-async-connect" option.
+	// "fluentd-async-connect" option was deprecated in Docker v20.10.0 and removed in v28.0.0.
+	// It was replaced with the "fluentd-async" option starting Docker v20.10.0.
+	// Docker v20.10.0 release notes: https://docs.docker.com/engine/release-notes/20.10/#logging-2
+	// Docker v28.0.0 release notes: https://docs.docker.com/engine/release-notes/28/#removed
+	// This change is not versioned and applies to all Docker client API versions.
+	// Hence, in order to preserve backwards-compatibility with Docker server versions older than v20.10.0,
+	// we need to continue using the older fluentd-async-connect option.
+	isAsyncCompatible, err := utils.Version(dockerServerVersion).Matches(">=20.10.0")
+	if err != nil {
+		logger.Error("Unable to determine whether the Docker server version is at least 20.10.0", logger.Fields{
+			field.TaskID:    task.GetID(),
+			field.Container: container.Name,
+			field.Error:     err,
+		})
+		return dockercontainer.LogConfig{}, errors.Wrapf(err,
+			"unable to determine whether the Docker server version is at least 20.10.0")
+	}
+	if isAsyncCompatible {
+		firelensConfig.Config[logDriverAsync] = strconv.FormatBool(cfg.FirelensAsyncEnabled.Enabled())
+	} else {
+		firelensConfig.Config[logDriverAsyncConnect] = strconv.FormatBool(cfg.FirelensAsyncEnabled.Enabled())
 	}
+
 	logger.Debug("Applying firelens log config for container", logger.Fields{
 		field.TaskID:    task.GetID(),
 		field.Container: container.Name,
-		"config":        logConfig,
+		"config":        firelensConfig,
 	})
-	return logConfig
+	return firelensConfig, nil
 }
 
 func (engine *DockerTaskEngine) startContainer(task *apitask.Task, container *apicontainer.Container) dockerapi.DockerContainerMetadata {
@@ -2882,3 +2954,104 @@ func (engine *DockerTaskEngine) getDockerID(task *apitask.Task, container *apico
 	}
 	return dockerContainer.DockerID, nil
 }
+
+// Sets CloudWatch Logs dual stack endpoint as "awslogs-endpoint" option in the logging config.
+// This is needed because awslogs driver that we consume from Docker does not support
+// an option to enable dual stack endpoints, so customers have no way to enable dual stack endpoints
+// that are needed in an IPv6-only environment.
+func (engine *DockerTaskEngine) setAWSLogsDualStackEndpoint(
+	task *apitask.Task, container *apicontainer.Container, hostConfig *dockercontainer.HostConfig,
+) {
+	// Helper function to populate common logger.Fields
+	withAdditionalLoggerFields := func(additionalFields logger.Fields) logger.Fields {
+		fields := logger.Fields{field.TaskARN: task.Arn, field.ContainerName: container.Name}
+		for k, v := range additionalFields {
+			fields[k] = v
+		}
+		return fields
+	}
+
+	// Do nothing if endpoint is already set
+	if hostConfig.LogConfig.Config[awsLogsEndpointKey] != "" {
+		logger.Info(
+			fmt.Sprintf(
+				"%s is already set in awslogs config, skip resolving dual stack CloudWatch Logs endpoint",
+				awsLogsEndpointKey),
+			withAdditionalLoggerFields(logger.Fields{}),
+		)
+		return
+	}
+
+	// Region is required to resolve endpoint
+	region := hostConfig.LogConfig.Config[awsLogsRegionKey]
+	if region == "" {
+		logger.Warn(
+			fmt.Sprintf(
+				"%s not found in awslogs config, skip resolving dual stack CloudWatch Logs endpoint",
+				awsLogsRegionKey),
+			withAdditionalLoggerFields(logger.Fields{}),
+		)
+		return
+	}
+
+	// Docker versions older than 18.09.0 do not support awslogs-endpoint
+	// option. So, skip endpoint resolution for those Docker versions.
+	dockerVersion, err := engine.Version()
+	if err != nil {
+		logger.Error("Failed to get Docker engine version. Skip resolving dual stack CloudWatch Logs endpoint.",
+			withAdditionalLoggerFields(logger.Fields{field.Error: err}))
+		return
+	}
+	const thresholdVersion = "18.09.0"
+	dockerVersionIsCompatible, err := utils.Version(dockerVersion).Matches(">=" + thresholdVersion)
+	if err != nil {
+		logger.Error("Failed to determine if docker version is high enough",
+			withAdditionalLoggerFields(logger.Fields{
+				field.Error:         err,
+				field.DockerVersion: dockerVersion,
+				"thresholdVersion":  thresholdVersion,
+			}))
+		return
+	}
+	if !dockerVersionIsCompatible {
+		logger.Warn(
+			fmt.Sprintf(
+				"Docker version does not support %s option. Skip resolving dual stack CloudWatch Logs endpoint.",
+				awsLogsEndpointKey),
+			withAdditionalLoggerFields(logger.Fields{
+				field.DockerVersion: dockerVersion,
+				"thresholdVersion":  thresholdVersion,
+			}),
+		)
+		return
+	}
+
+	// Resolve the endpoint
+	endpoint, err := getAWSLogsDualStackEndpoint(region)
+	if err != nil {
+		logger.Error(
+			"Failed to get CloudWatch Logs dual stack endpoint. Skipping setting it.",
+			withAdditionalLoggerFields(logger.Fields{field.Region: region, field.Error: err}))
+		return
+	}
+
+	logger.Info("Resolved CloudWatch Logs dual stack endpoint",
+		withAdditionalLoggerFields(logger.Fields{
+			field.Endpoint: endpoint,
+			field.Region:   region,
+		}))
+	hostConfig.LogConfig.Config[awsLogsEndpointKey] = endpoint
+}
+
+// Returns CloudWatch Logs dual stack endpoint for the given region.
+func getAWSLogsDualStackEndpoint(region string) (string, error) {
+	endpoint, err := cloudwatchlogs.NewDefaultEndpointResolverV2().ResolveEndpoint(context.TODO(),
+		cloudwatchlogs.EndpointParameters{
+			UseDualStack: ptr.Bool(true),
+			Region:       ptr.String(region),
+		})
+	if err != nil {
+		return "", fmt.Errorf("failed to resolve dual stack CloudWatch Logs endpoint for region '%s': %w", region, err)
+	}
+	return endpoint.URI.String(), nil
+}