diff --git a/daemon/logger/awslogs/cloudwatchlogs.go b/daemon/logger/awslogs/cloudwatchlogs.go
index dce0940ac7..0ed42e6651 100644
--- a/daemon/logger/awslogs/cloudwatchlogs.go
+++ b/daemon/logger/awslogs/cloudwatchlogs.go
@@ -42,6 +42,7 @@ const (
 	credentialsEndpointKey = "awslogs-credentials-endpoint"
 	forceFlushIntervalKey  = "awslogs-force-flush-interval-seconds"
 	maxBufferedEventsKey   = "awslogs-max-buffered-events"
+	logFormatKey           = "awslogs-format"
 
 	defaultForceFlushInterval = 5 * time.Second
 	defaultMaxBufferedEvents  = 4096
@@ -66,6 +67,10 @@ const (
 	credentialsEndpoint = "http://169.254.170.2"
 
 	userAgentHeader = "User-Agent"
+
+	// See: https://docs.aws.amazon.com/AmazonCloudWatch/latest/monitoring/CloudWatch_Embedded_Metric_Format_Specification.html
+	logsFormatHeader = "x-amzn-logs-format"
+	jsonEmfLogFormat = "json/emf"
 )
 
 type logStream struct {
@@ -404,6 +409,16 @@ func newAWSLogsClient(info logger.Info) (api, error) {
 					dockerversion.Version, runtime.GOOS, currentAgent))
 		},
 	})
+
+	if info.Config[logFormatKey] != "" {
+		client.Handlers.Build.PushBackNamed(request.NamedHandler{
+			Name: "LogFormatHeaderHandler",
+			Fn: func(req *request.Request) {
+				req.HTTPRequest.Header.Set(logsFormatHeader, info.Config[logFormatKey])
+			},
+		})
+	}
+
 	return client, nil
 }
 
@@ -755,6 +770,7 @@ func ValidateLogOpt(cfg map[string]string) error {
 		case credentialsEndpointKey:
 		case forceFlushIntervalKey:
 		case maxBufferedEventsKey:
+		case logFormatKey:
 		default:
 			return fmt.Errorf("unknown log opt '%s' for %s log driver", key, name)
 		}
@@ -782,6 +798,17 @@ func ValidateLogOpt(cfg map[string]string) error {
 	if datetimeFormatKeyExists && multilinePatternKeyExists {
 		return fmt.Errorf("you cannot configure log opt '%s' and '%s' at the same time", datetimeFormatKey, multilinePatternKey)
 	}
+
+	if cfg[logFormatKey] != "" {
+		// For now, only the "json/emf" log format is supported
+		if cfg[logFormatKey] != jsonEmfLogFormat {
+			return fmt.Errorf("unsupported log format '%s'", cfg[logFormatKey])
+		}
+		if datetimeFormatKeyExists || multilinePatternKeyExists {
+			return fmt.Errorf("you cannot configure log opt '%s' or '%s' when log opt '%s' is set to '%s'", datetimeFormatKey, multilinePatternKey, logFormatKey, jsonEmfLogFormat)
+		}
+	}
+
 	return nil
 }
 
diff --git a/daemon/logger/awslogs/cloudwatchlogs_test.go b/daemon/logger/awslogs/cloudwatchlogs_test.go
index ed1465b4d4..28b521d1fa 100644
--- a/daemon/logger/awslogs/cloudwatchlogs_test.go
+++ b/daemon/logger/awslogs/cloudwatchlogs_test.go
@@ -147,6 +147,48 @@ func TestNewAWSLogsClientUserAgentHandler(t *testing.T) {
 	}
 }
 
+func TestNewAWSLogsClientLogFormatHeaderHandler(t *testing.T) {
+	tests := []struct {
+		logFormat           string
+		expectedHeaderValue string
+	}{
+		{
+			logFormat:           jsonEmfLogFormat,
+			expectedHeaderValue: "json/emf",
+		},
+		{
+			logFormat:           "",
+			expectedHeaderValue: "",
+		},
+	}
+	for _, tc := range tests {
+		t.Run(tc.logFormat, func(t *testing.T) {
+			info := logger.Info{
+				Config: map[string]string{
+					regionKey:    "us-east-1",
+					logFormatKey: tc.logFormat,
+				},
+			}
+
+			client, err := newAWSLogsClient(info)
+			assert.NilError(t, err)
+
+			realClient, ok := client.(*cloudwatchlogs.CloudWatchLogs)
+			assert.Check(t, ok, "Could not cast client to cloudwatchlogs.CloudWatchLogs")
+
+			buildHandlerList := realClient.Handlers.Build
+			request := &request.Request{
+				HTTPRequest: &http.Request{
+					Header: http.Header{},
+				},
+			}
+			buildHandlerList.Run(request)
+			logFormatHeaderVal := request.HTTPRequest.Header.Get("x-amzn-logs-format")
+			assert.Equal(t, tc.expectedHeaderValue, logFormatHeaderVal)
+		})
+	}
+}
+
 func TestNewAWSLogsClientAWSLogsEndpoint(t *testing.T) {
 	endpoint := "mock-endpoint"
 	info := logger.Info{
@@ -1559,6 +1601,43 @@ func TestValidateLogOptionsMaxBufferedEvents(t *testing.T) {
 	}
 }
 
+func TestValidateLogOptionsFormat(t *testing.T) {
+	tests := []struct {
+		format           string
+		multiLinePattern string
+		datetimeFormat   string
+		expErrMsg        string
+	}{
+		{"json/emf", "", "", ""},
+		{"random", "", "", "unsupported log format 'random'"},
+		{"", "", "", ""},
+		{"json/emf", "---", "", "you cannot configure log opt 'awslogs-datetime-format' or 'awslogs-multiline-pattern' when log opt 'awslogs-format' is set to 'json/emf'"},
+		{"json/emf", "", "yyyy-dd-mm", "you cannot configure log opt 'awslogs-datetime-format' or 'awslogs-multiline-pattern' when log opt 'awslogs-format' is set to 'json/emf'"},
+	}
+
+	for i, tc := range tests {
+		t.Run(fmt.Sprintf("%d/%s", i, tc.format), func(t *testing.T) {
+			cfg := map[string]string{
+				logGroupKey:  groupName,
+				logFormatKey: tc.format,
+			}
+			if tc.multiLinePattern != "" {
+				cfg[multilinePatternKey] = tc.multiLinePattern
+			}
+			if tc.datetimeFormat != "" {
+				cfg[datetimeFormatKey] = tc.datetimeFormat
+			}
+
+			err := ValidateLogOpt(cfg)
+			if tc.expErrMsg != "" {
+				assert.Error(t, err, tc.expErrMsg)
+			} else {
+				assert.NilError(t, err)
+			}
+		})
+	}
+}
+
 func TestCreateTagSuccess(t *testing.T) {
 	mockClient := newMockClient()
 	info := logger.Info{