Determine the fluentd async option based on Docker server version

Docker's fluentd log driver supports passing an "async" option.
The "fluentd-async-connect" option was deprecated in Docker v20.10.0 and completely removed in v28.0.0.
Reference: https://docs.docker.com/engine/deprecated/#fluentd-async-connect-log-opt
Functionally, both options are equivalent, see: moby/moby#39086.

Since Docker's refactor of the fluentd async option was not versioned and affects all API versions,
we need to determine the appropriate option using the Docker Server version, in order to be backwards compatible.

Author: singholt

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

@@ -97,6 +97,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 +1875,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,
@@ -2108,29 +2137,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 {

agent/engine/docker_task_engine_test.go

@@ -115,6 +115,7 @@ const (
 	serviceConnectContainerName = "service-connect"
 	mediaTypeManifestV2         = "application/vnd.docker.distribution.manifest.v2+json"
 	defaultIfname               = "eth0"
+	testDockerServerVersion     = "25.0.8"
 )
 
 var (
@@ -3321,7 +3322,7 @@ func TestCreateContainerAddFirelensLogDriverConfig(t *testing.T) {
 		expectedLogConfigType          string
 		expectedLogConfigTag           string
 		expectedLogConfigFluentAddress string
-		expectedFluentdAsyncConnect    string
+		expectedFluentdAsync           string
 		expectedSubSecondPrecision     string
 		expectedBufferLimit            string
 		expectedIPAddress              string
@@ -3333,7 +3334,7 @@ func TestCreateContainerAddFirelensLogDriverConfig(t *testing.T) {
 			enableServiceConnect:           false,
 			expectedLogConfigType:          logDriverTypeFluentd,
 			expectedLogConfigTag:           taskName + "-firelens-" + taskID,
-			expectedFluentdAsyncConnect:    strconv.FormatBool(true),
+			expectedFluentdAsync:           strconv.FormatBool(true),
 			expectedSubSecondPrecision:     strconv.FormatBool(true),
 			expectedBufferLimit:            "10000",
 			expectedLogConfigFluentAddress: socketPathPrefix + filepath.Join(defaultConfig.DataDirOnHost, dataLogDriverPath, taskID, dataLogDriverSocketPath),
@@ -3346,7 +3347,7 @@ func TestCreateContainerAddFirelensLogDriverConfig(t *testing.T) {
 			enableServiceConnect:           false,
 			expectedLogConfigType:          logDriverTypeFluentd,
 			expectedLogConfigTag:           taskName + "-firelens-" + taskID,
-			expectedFluentdAsyncConnect:    strconv.FormatBool(true),
+			expectedFluentdAsync:           strconv.FormatBool(true),
 			expectedSubSecondPrecision:     strconv.FormatBool(true),
 			expectedBufferLimit:            "10000",
 			expectedLogConfigFluentAddress: socketPathPrefix + filepath.Join(defaultConfig.DataDirOnHost, dataLogDriverPath, taskID, dataLogDriverSocketPath),
@@ -3359,7 +3360,7 @@ func TestCreateContainerAddFirelensLogDriverConfig(t *testing.T) {
 			enableServiceConnect:           true,
 			expectedLogConfigType:          logDriverTypeFluentd,
 			expectedLogConfigTag:           taskName + "-firelens-" + taskID,
-			expectedFluentdAsyncConnect:    strconv.FormatBool(true),
+			expectedFluentdAsync:           strconv.FormatBool(true),
 			expectedSubSecondPrecision:     strconv.FormatBool(true),
 			expectedBufferLimit:            "10000",
 			expectedLogConfigFluentAddress: socketPathPrefix + filepath.Join(defaultConfig.DataDirOnHost, dataLogDriverPath, taskID, dataLogDriverSocketPath),
@@ -3372,7 +3373,7 @@ func TestCreateContainerAddFirelensLogDriverConfig(t *testing.T) {
 			enableServiceConnect:           false,
 			expectedLogConfigType:          logDriverTypeFluentd,
 			expectedLogConfigTag:           taskName + "-firelens-" + taskID,
-			expectedFluentdAsyncConnect:    strconv.FormatBool(true),
+			expectedFluentdAsync:           strconv.FormatBool(true),
 			expectedSubSecondPrecision:     strconv.FormatBool(true),
 			expectedBufferLimit:            "",
 			expectedLogConfigFluentAddress: socketPathPrefix + filepath.Join(defaultConfig.DataDirOnHost, dataLogDriverPath, taskID, dataLogDriverSocketPath),
@@ -3388,6 +3389,7 @@ func TestCreateContainerAddFirelensLogDriverConfig(t *testing.T) {
 			ctrl, client, _, taskEngine, _, _, _, serviceConnectManager := mocks(t, ctx, &defaultConfig)
 			defer ctrl.Finish()
 
+			client.EXPECT().Version(gomock.Any(), gomock.Any()).Return(testDockerServerVersion, nil).Times(1)
 			client.EXPECT().APIVersion().Return(defaultDockerClientAPIVersion, nil).AnyTimes()
 			if tc.enableServiceConnect {
 				serviceConnectManager.EXPECT().AugmentTaskContainer(gomock.Any(), gomock.Any(), gomock.Any()).Return(nil).Times(1)
@@ -3401,7 +3403,7 @@ func TestCreateContainerAddFirelensLogDriverConfig(t *testing.T) {
 					assert.Equal(t, tc.expectedLogConfigType, hostConfig.LogConfig.Type)
 					assert.Equal(t, tc.expectedLogConfigTag, hostConfig.LogConfig.Config["tag"])
 					assert.Equal(t, tc.expectedLogConfigFluentAddress, hostConfig.LogConfig.Config["fluentd-address"])
-					assert.Equal(t, tc.expectedFluentdAsyncConnect, hostConfig.LogConfig.Config["fluentd-async-connect"])
+					assert.Equal(t, tc.expectedFluentdAsync, hostConfig.LogConfig.Config["fluentd-async"])
 					assert.Equal(t, tc.expectedSubSecondPrecision, hostConfig.LogConfig.Config["fluentd-sub-second-precision"])
 					assert.Equal(t, tc.expectedBufferLimit, hostConfig.LogConfig.Config["fluentd-buffer-limit"])
 					assert.Contains(t, config.Env, tc.expectedIPAddress)
@@ -5211,8 +5213,102 @@ func TestGetFirelensConfigWithAsyncEnabledConfigOption(t *testing.T) {
 			cfg := &config.Config{
 				FirelensAsyncEnabled: asyncEnabled,
 			}
-			logConfig := getFirelensLogConfig(task, appContainer, rawHostConfigInput, cfg)
-			assert.Equal(t, tc.isFirelensAsyncEnabled, logConfig.Config[logDriverAsyncConnect])
+			logConfig, err := getFirelensLogConfig(task, appContainer, rawHostConfigInput, cfg,
+				testDockerServerVersion)
+			assert.NoError(t, err)
+			assert.Equal(t, tc.isFirelensAsyncEnabled, logConfig.Config[logDriverAsync])
+		})
+	}
+}
+
+// TestGetFirelensConfigBasedOnDockerServerVersion validates that GetFirelensConfig populates the correct fluentd async
+// option depending on the Docker Server version provided.
+func TestGetFirelensConfigBasedOnDockerServerVersion(t *testing.T) {
+	rawHostConfigInput := &dockercontainer.HostConfig{
+		LogConfig: dockercontainer.LogConfig{
+			Type: "awsfirelens",
+			Config: map[string]string{
+				"log-driver-buffer-limit": "10000",
+			},
+		},
+	}
+
+	rawHostConfig, err := json.Marshal(&rawHostConfigInput)
+	require.NoError(t, err)
+	hostConfig := func() *string {
+		s := string(rawHostConfig)
+		return &s
+	}()
+
+	appContainer := &apicontainer.Container{
+		Name: "app",
+		DockerConfig: apicontainer.DockerConfig{
+			HostConfig: hostConfig,
+		},
+	}
+
+	firelensContainer := &apicontainer.Container{
+		Name: "firelens",
+		FirelensConfig: &apicontainer.FirelensConfig{
+			Type: "fluentbit",
+		},
+	}
+
+	task := &apitask.Task{
+		Arn: "arn:aws:ecs:region:account-id:task/task-id",
+		Containers: []*apicontainer.Container{
+			appContainer,
+			firelensContainer,
+		},
+	}
+
+	testCases := []struct {
+		name                string
+		dockerServerVersion string
+		expectError         bool
+		includedAsyncOption string
+		excludedAsyncOption string
+	}{
+		{
+			name:                "use fluentd-async",
+			dockerServerVersion: testDockerServerVersion,
+			expectError:         false,
+			includedAsyncOption: logDriverAsync,
+			excludedAsyncOption: logDriverAsyncConnect,
+		},
+		{
+			name:                "use fluentd-async-connect",
+			dockerServerVersion: "19.03.13-ce",
+			expectError:         false,
+			includedAsyncOption: logDriverAsyncConnect,
+			excludedAsyncOption: logDriverAsync,
+		},
+		{
+			name:                "error while determining Docker server version",
+			dockerServerVersion: "",
+			expectError:         true,
+			includedAsyncOption: "",
+			excludedAsyncOption: "",
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+
+			cfg := &config.Config{
+				FirelensAsyncEnabled: config.BooleanDefaultTrue{Value: config.ExplicitlyEnabled},
+			}
+			logConfig, configError := getFirelensLogConfig(task, appContainer, rawHostConfigInput, cfg,
+				tc.dockerServerVersion)
+			assert.Equal(t, tc.expectError, configError != nil)
+			// Verify whether the logConfig.Config map contains the right fluentd async key and value
+			if tc.includedAsyncOption != "" {
+				assert.Contains(t, logConfig.Config, tc.includedAsyncOption)
+				assert.Equal(t, "true", logConfig.Config[tc.includedAsyncOption])
+			} else {
+				assert.NotContains(t, logConfig.Config, tc.includedAsyncOption)
+			}
+			assert.NotContains(t, logConfig.Config, tc.excludedAsyncOption)
 		})
 	}
 }