Enable support for Embedded Metric Format

README.md

@@ -25,6 +25,7 @@ Run `make` to build `./bin/cloudwatch.so`. Then use with Fluent Bit:
 * `log_stream_name`: The name of the CloudWatch Log Stream that you want log records sent to.
 * `log_stream_prefix`: Prefix for the Log Stream name. The tag is appended to the prefix to construct the full log stream name. Not compatible with the `log_stream_name` option.  
 * `log_key`: By default, the whole log record will be sent to CloudWatch. If you specify a key name with this option, then only the value of that key will be sent to CloudWatch. For example, if you are using the Fluentd Docker log driver, you can specify `log_key log` and only the log message will be sent to CloudWatch.
+* `log_format`: An optional parameter that can be used to tell CloudWatch the format of the data. A value of `json/emf` enables CloudWatch to extract custom metrics embedded in a JSON payload. See the [Embedded Metric Format](https://docs.aws.amazon.com/AmazonCloudWatch/latest/monitoring/CloudWatch_Embedded_Metric_Format_Specification.html).
 * `role_arn`: ARN of an IAM role to assume (for cross account access).
 * `auto_create_group`: Automatically create the log group. Valid values are "true" or "false" (case insensitive). Defaults to false.
 * `endpoint`: Specify a custom endpoint for the CloudWatch Logs API.

cloudwatch/cloudwatch.go

@@ -105,6 +105,7 @@ type OutputPluginConfig struct {
 	CWEndpoint       string
 	CredsEndpoint    string
 	PluginInstanceID int
+	LogFormat        string
 }
 
 // Validate checks the configuration input for an OutputPlugin instances
@@ -132,7 +133,7 @@ func NewOutputPlugin(config OutputPluginConfig) (*OutputPlugin, error) {
 		return nil, err
 	}
 
-	client := newCloudWatchLogsClient(config.RoleARN, sess, config.CWEndpoint, config.CredsEndpoint)
+	client := newCloudWatchLogsClient(config.RoleARN, sess, config.CWEndpoint, config.CredsEndpoint, config.LogFormat)
 
 	timer, err := plugins.NewTimeout(func(d time.Duration) {
 		logrus.Errorf("[cloudwatch %d] timeout threshold reached: Failed to send logs for %s\n", config.PluginInstanceID, d.String())
@@ -157,7 +158,7 @@ func NewOutputPlugin(config OutputPluginConfig) (*OutputPlugin, error) {
 	}, nil
 }
 
-func newCloudWatchLogsClient(roleARN string, sess *session.Session, endpoint string, credsEndpoint string) *cloudwatchlogs.CloudWatchLogs {
+func newCloudWatchLogsClient(roleARN string, sess *session.Session, endpoint string, credsEndpoint string, logFormat string) *cloudwatchlogs.CloudWatchLogs {
 	svcConfig := &aws.Config{}
 	if endpoint != "" {
 		defaultResolver := endpoints.DefaultResolver()
@@ -185,6 +186,11 @@ func newCloudWatchLogsClient(roleARN string, sess *session.Session, endpoint str
 
 	client := cloudwatchlogs.New(sess, svcConfig)
 	client.Handlers.Build.PushBackNamed(plugins.CustomUserAgentHandler())
+
+	if logFormat != "" {
+		client.Handlers.Build.PushBackNamed(LogFormatHandler(logFormat))
+	}
+
 	return client
 }
 

cloudwatch/handlers.go

@@ -0,0 +1,29 @@
+// Copyright 2019 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+//
+// Licensed under the Apache License, Version 2.0 (the "License"). You may
+// not use this file except in compliance with the License. A copy of the
+// License is located at
+//
+//	http://aws.amazon.com/apache2.0/
+//
+// or in the "license" file accompanying this file. This file 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 cloudwatch
+
+import "github.com/aws/aws-sdk-go/aws/request"
+
+const logFormatHeader = "x-amzn-logs-format"
+
+// LogFormatHandler returns an http request handler that sets an HTTP header used to 
+// indicate the format of the logs being sent
+func LogFormatHandler(format string) request.NamedHandler {
+	return request.NamedHandler{
+		Name: "LogFormatHandler",
+		Fn: func(req *request.Request) {
+			req.HTTPRequest.Header.Set("x-amzn-logs-format", format)
+		},
+	}
+}

cloudwatch/handlers_test.go

@@ -0,0 +1,37 @@
+// Copyright 2019 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+//
+// Licensed under the Apache License, Version 2.0 (the "License"). You may
+// not use this file except in compliance with the License. A copy of the
+// License is located at
+//
+//	http://aws.amazon.com/apache2.0/
+//
+// or in the "license" file accompanying this file. This file 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 cloudwatch
+
+import (
+	"net/http"
+	"testing"
+
+	"github.com/aws/aws-sdk-go/aws/request"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestLogFormatHandler(t *testing.T) {
+	httpReq, _ := http.NewRequest("POST", "", nil)
+	r := &request.Request{
+		HTTPRequest: httpReq,
+		Body:        nil,
+	}
+	r.SetBufferBody([]byte{})
+
+	handler := LogFormatHandler("json/emf")
+	handler.Fn(r)
+
+	header := r.HTTPRequest.Header.Get("x-amzn-logs-format")
+	assert.Equal(t, "json/emf", header)
+}

fluent-bit-cloudwatch.go

@@ -85,6 +85,8 @@ func getConfiguration(ctx unsafe.Pointer, pluginID int) cloudwatch.OutputPluginC
 	logrus.Infof("[cloudwatch %d] plugin parameter endpoint = '%s'\n", pluginID, config.CWEndpoint)
 	config.CredsEndpoint = output.FLBPluginConfigKey(ctx, "credentials_endpoint")
 	logrus.Infof("[cloudwatch %d] plugin parameter credentials_endpoint = %s\n", pluginID, config.CredsEndpoint)
+	config.LogFormat = output.FLBPluginConfigKey(ctx, "log_format")
+	logrus.Infof("[cloudwatch %d] plugin parameter log_format = '%s'\n", pluginID, config.LogFormat)
 
 	return config
 }