Sink Declarations

Sink declarations specify where and in what form logs, events, and metrics should be sent to various AWS services. The following sections describe configurations for the built-in sink types that are available in Amazon Kinesis Agent for Microsoft Windows. Because Kinesis Agent for Windows is extensible, you can add custom sink types. Each sink type typically requires unique key-value pairs in the configuration declarations that are relevant for that sink type.

All sink declarations can contain the following key-value pairs:

For examples of complete configuration files that use various kinds of sinks, see Streaming from the Windows Application Event Log to Sinks.

KinesisStream Sink Configuration

The KinesisStream sink type streams log records and events to the Kinesis Data Streams service. Typically, data that is streamed to Kinesis Data Streams is processed by one or more custom applications that execute using various AWS services. Data is streamed to a named stream that is configured using Kinesis Data Streams. For more information, see the Amazon Kinesis Data Streams Developer Guide.

The following is an example Kinesis Data Streams sink declaration:

{
    "Id": "TestKinesisStreamSink",
    "SinkType": "KinesisStream",
    "StreamName": "MyTestStream",
    "Region": "us-west-2"
}
   

All KinesisStream sink declarations can provide the following additional key-value pairs:

The default BufferInterval for this sink type is 1 second, and the default BufferSize is 500 records.

KinesisFirehose Sink Configuration

The KinesisFirehose sink type streams log records and events to the Firehose service. Firehose delivers the streamed data to other services for storage. Typically the stored data is then analyzed in subsequent stages of the data pipeline. Data is streamed to a named delivery stream that is configured using Firehose. For more information, see the Amazon Data Firehose Developer Guide.

The following is an example Firehose sink declaration:

{
   "Id": "TestKinesisFirehoseSink",
   "SinkType": "KinesisFirehose",
   "StreamName": "MyTestFirehoseDeliveryStream",
   "Region": "us-east-1",
   "CombineRecords": "true"
}
   

All KinesisFirehose sink declarations can provide the following additional key-value pairs:

The default BufferInterval for this sink type is 1 second, and the default BufferSize is 500 records.

CloudWatch Sink Configuration

The CloudWatch sink type streams metrics to the CloudWatch service. You can view the metrics in the AWS Management Console. For more information, see the Amazon CloudWatch User Guide.

The following is an example CloudWatch sink declaration:

{
   "Id": "CloudWatchSink",
   "SinkType": "CloudWatch"
}
   

All CloudWatch sink declarations can provide the following additional key-value pairs:

CloudWatchLogs Sink Configuration

The CloudWatchLogs sink type streams log records and events to Amazon CloudWatch Logs. You can view logs in the AWS Management Console, or process them via additional stages of a data pipeline. Data is streamed to a named log stream that is configured in CloudWatch Logs. Log streams are organized into named log groups. For more information, see the Amazon CloudWatch Logs User Guide.

The following is an example CloudWatch Logs sink declaration:

{
   "Id": "MyCloudWatchLogsSink",
   "SinkType": "CloudWatchLogs",
   "BufferInterval": "60",
   "BufferSize": "100",
   "Region": "us-west-2",
   "LogGroup": "MyTestLogGroup",
   "LogStream": "MyTestStream"
}

All CloudWatchLogs sink declarations must provide the following additional key-value pairs:

The default BufferInterval for this sink type is 1 second, and the default BufferSize is 500 records. The maximum buffer size is 10,000 records.

Local FileSystem Sink Configuration

The sink type FileSystem saves log and event records to a file on the local file system instead of streaming them to AWS services. FileSystem sinks are useful for testing and diagnostics. For example, you can use this sink type to examine records before sending them to AWS.

With FileSystem sinks, you can also use configuration parameters to simulate batching, throttling, and retry-on-error to mimic the behavior of actual AWS sinks.

All records from all sources connected to a FileSystem sink are saved to the single file specified as FilePath. If FilePath is not specified, records are saved to a file named SinkId.txt in the %TEMP% directory, which is usually C:\Users\UserName\AppData\Local\Temp, where SinkId is the unique identifier of the sink and UserName is the Windows user name of the active user.

This sink type supports text decoration attributes. For more information, see Configuring Sink Decorations.

An example FileSystem sink type configuration is shown in the following example.

{
	   "Id": "LocalFileSink",
	   "SinkType": "FileSystem",
	   "FilePath": "C:\\ProgramData\\Amazon\\local_sink.txt",
	   "Format": "json",
	   "TextDecoration": "",
	   "ObjectDecoration": ""
}

The FileSystem configuration consists of the following key-value pairs.

Advanced Usage – Record Throttling and Failure Simulation

FileSystem can mimic the behavior of AWS sinks by simulating record throttling. You can use the following key-value pairs to specify record throttling and failure simulation attributes.

By acquiring a lock on the destination file and preventing writes to it, you can use FileSystem sinks to simulate and examine the behavior of AWS sinks when the network fails.

The following example shows a FileSystem configuration with simulation attributes.

{
	   "Id": "LocalFileSink",
	   "SinkType": "FileSystem",
	   "FilePath": "C:\\ProgramData\\Amazon\\local_sink.txt",
	   "TextDecoration": "",
	   "RequestsPerSecond": "100",
    "BufferSize": "10",
    "MaxBatchSize": "1024"
}

The maximum record rate limit is a function of BufferSize, which determines the maximum number of records per request, and RequestsPerSecond. You can calculate the record rate limit per second using the following formula.

RecordRate = BufferSize * RequestsPerSecond

Given configuration values in the example above, there is a maximum record rate of 1000 records per second.

Sink Security Configuration

Configuring Authentication

For Kinesis Agent for Windows to stream logs, events, and metrics to AWS services, access must be authenticated. There are several ways to provide authentication for Kinesis Agent for Windows. How you do it depends on the situation where Kinesis Agent for Windows is executing and the specific security requirements for a particular organization.

Although it is possible to directly provide access keys and secret keys in the sink declarations, this approach is discouraged because the declarations are not encrypted.

Configuring Authorization

Attach the appropriate policies that follow to the IAM user or role that Kinesis Agent for Windows will use to stream data to AWS services:

Kinesis Data Streams

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "VisualEditor0",
            "Effect": "Allow",
            "Action": [
                "kinesis:PutRecord",
                "kinesis:PutRecords"
            ],
            "Resource": "arn:aws:kinesis:*:*:stream/*"
        }
    ]
}     
     

To limit authorization to a specific Region, account, or stream name, replace the appropriate asterisks in the ARN with specific values. For more information, see "Amazon Resource Names (ARNs) for Kinesis Data Streams" in Controlling Access to Amazon Kinesis Data Streams Resources Using IAM.

Firehose

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "VisualEditor1",
            "Effect": "Allow",
            "Action": [
                "firehose:PutRecord",
                "firehose:PutRecordBatch"
            ],
            "Resource": "arn:aws:firehose:*:*:deliverystream/*"
        }
    ]
}      
     

To limit authorization to a specific Region, account, or delivery stream name, replace the appropriate asterisks in the ARN with specific values. For more information, see Controlling Access with Amazon Kinesis Data Firehose in the Amazon Data Firehose Developer Guide.

CloudWatch

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "VisualEditor2",
            "Effect": "Allow",
            "Action": "cloudwatch:PutMetricData",
            "Resource": "*"
        }
    ]
}

For more information, see Overview of Managing Access Permissions to Your CloudWatch Resources in the Amazon CloudWatch Logs User Guide.

CloudWatch Logs with an Existing Log Group and Log Stream

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "VisualEditor3",
            "Effect": "Allow",
            "Action": [
                "logs:DescribeLogGroups",
                "logs:DescribeLogStreams",
                "logs:PutLogEvents"
            ],
            "Resource": "arn:aws:logs:*:*:log-group:*"
        },
        {
            "Sid": "VisualEditor4",
            "Effect": "Allow",
            "Action": "logs:PutLogEvents",
            "Resource": "arn:aws:logs:*:*:log-group:*:*:*"
        }
    ]
}      
     

To restrict access to a specific Region, account, log group, or log stream, replace the appropriate asterisks in the ARNs with appropriate values. For more information, see Overview of Managing Access Permissions to Your CloudWatch Logs Resources in the Amazon CloudWatch Logs User Guide.

CloudWatch Logs with Extra Permissions for Kinesis Agent for Windows to Create Log Groups and Log Streams

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "VisualEditor5",
            "Effect": "Allow",
            "Action": [
                "logs:CreateLogStream",
                "logs:DescribeLogGroups",
                "logs:DescribeLogStreams",
                "logs:PutLogEvents"
            ],
            "Resource": "arn:aws:logs:*:*:log-group:*"
        },
        {
            "Sid": "VisualEditor6",
            "Effect": "Allow",
            "Action": "logs:PutLogEvents",
            "Resource": "arn:aws:logs:*:*:log-group:*:*:*"
        },
        {
            "Sid": "VisualEditor7",
            "Effect": "Allow",
            "Action": "logs:CreateLogGroup",
            "Resource": "*"
        }
    ]
}
     

To restrict access to a specific Region, account, log group, or log stream, replace the appropriate asterisks in the ARNs with appropriate values. For more information, see Overview of Managing Access Permissions to Your CloudWatch Logs Resources in the Amazon CloudWatch Logs User Guide.

Permissions Required for EC2 Tag Variable Expansion

Using variable expansion with the ec2tag variable prefix requires the ec2:Describe* permission.

{
   "Version": "2012-10-17",
   "Statement": [{
      "Sid": "VisualEditor8",
      "Effect": "Allow",
      "Action": "ec2:Describe*",
      "Resource": "*"
    }
   ]
}     

Configuring ProfileRefreshingAWSCredentialProvider to Refresh AWS Credentials

If you use AWS Systems Manager for hybrid environments to manage AWS credentials, Systems Manager rotates session credentials in c:\Windows\System32\config\systemprofile\.aws\credentials. For more information about Systems Manager for hybrid environments, see Setting up AWS Systems Manager for hybrid environments in the AWS Systems Manager User Guide.

Because the AWS .net SDK does not pick up new credentials automatically, we provide the ProfileRefreshingAWSCredentialProvider plug-in to refresh credentials.

You can use the CredentialRef attribute of any AWS sync configuration to reference a Credentials definition where the CredentialType attribute is set to ProfileRefreshingAWSCredentialProvider as shown in the following example.

{
    "Sinks": [{
		      "Id": "myCloudWatchLogsSink",
		      "SinkType": "CloudWatchLogs",
		      "CredentialRef": "ssmcred",
		      "Region": "us-west-2",
		      "LogGroup": "myLogGroup",
		      "LogStream": "myLogStream"
    }],
    "Credentials": [{
        "Id": "ssmcred",
        "CredentialType": "ProfileRefreshingAWSCredentialProvider",
        "Profile": "default",
        "FilePath": "%USERPROFILE%//.aws//credentials",
        "RefreshingInterval": 300
    }]
}

A credential definition consists of the following attributes as key-value pairs.

Configuring Sink Decorations

Sink declarations can optionally include key-value pairs that specify additional data to stream to various AWS services to enhance the records gathered from the source.

If the source type of the source connected to the sink is DirectorySource, then the sink can use three additional variables:

These variables are useful when you use a source that gathers log records from multiple files connected to a sink that streams all the records to a single stream. Injecting the values of these variables into the streaming records enables downstream analytics in the data pipeline to order the records by file and by location within each file.

Tips for Writing Expressions

An expression can be any of the following:

Special Characters

Two backslashes are required to escape special characters.

Nesting

Function invocations can be nested, as shown in the following example.

format(date(2018, 11, 28), 'MMddyyyy')

Variables

There are three types of variables: local, meta, and global.

Built-in Functions

Kinesis Agent for Windows supports the following built-in functions. If any of the arguments are NULL and the function is not designed to handle NULL, a NULL object is returned.

//string functions
int length(string input)
string lower(string input)
string lpad(string input, int size, string padstring)
string ltrim(string input)
string rpad(string input, int size, string padstring)
string rtrim(string input)
string substr(string input, int start)
string substr(string input, int start, int length)
string trim(string input)
string upper(string str)

//regular expression functions
string regexp_extract(string input, string pattern)
string regexp_extract(string input, string pattern, int group)

//date functions
DateTime date(int year, int month, int day)
DateTime date(int year, int month, int day, int hour, int minute, int second)
DateTime date(int year, int month, int day, int hour, int minute, int second, int millisecond)

//conversion functions
int? parse_int(string input)
decimal? parse_decimal(string input)
DateTime? parse_date(string input, string format)
string format(object o, string format)

//coalesce functions
object coalesce(object obj1, object obj2)
object coalesce(object obj1, object obj2, object obj3)
object coalesce(object obj1, object obj2, object obj3, object obj4)
object coalesce(object obj1, object obj2, object obj3, object obj4, object obj5) 
object coalesce(object obj1, object obj2, object obj3, object obj4, object obj5, object obj6)

Configuring Sink Variable Substitutions

The KinesisStream, KinesisFirehose, and CloudWatchLogs sink declarations require either a LogStream or StreamName key-value pair. The value of these key-values can contain variable references that are automatically resolved by Kinesis Agent for Windows. For CloudWatchLogs, the LogGroup key-value pair is also required and can contain variables references that are automatically resolved by Kinesis Agent for Windows. The variables are specified using the template {prefix:variablename} where prefix: is optional. The supported prefixes are as follows:

If the prefix isn't specified, if there is an environment variable with the same name as variablename, the variable reference is resolved to the value of the environment variable. Otherwise, if variablename is instance_id or hostname, the variable reference is resolved to the value of the EC2 metadata of the same name. Otherwise, the variable reference is not resolved.

The following are examples of valid key-value pairs using variable references:

"LogStream": "LogStream_{instance_id}"
"LogStream": "LogStream_{hostname}"
"LogStream": "LogStream_{ec2:local-hostname}"
"LogStream": "LogStream_{computername}"
"LogStream": "LogStream_{env:computername}"
   

The CloudWatchLogs sink declarations support a special format timestamp variable that allows the timestamp of the original log or event record from the source to alter the name of the log stream. The format is {timestamp:timeformat}. See the following example:

"LogStream": "LogStream_{timestamp:yyyyMMdd}"

If the log or event record was generated on June 5, 2017, the value of the LogStream key-value pair in the previous example would resolve to "LogStream_20170605".

If authorized, the CloudWatchLogs sink type can automatically create new log streams when required based on the generated names. You cannot do this for other sink types because they require additional configuration beyond the name of the stream.

There are special variable substitutions that occur in text and object decoration. For more information, see Configuring Sink Decorations.

Configuring Sink Queuing

The KinesisStream, KinesisFirehose, and CloudWatchLogs sink declarations can optionally enable queuing of records that have failed to stream to the AWS service associated with those sink types due to transient connectivity issues. To enable queuing and automatic streaming retries when connectivity is restored, use the following key-value pairs in the sink declarations:

Configuring a Proxy for Sinks

To configure a proxy for all the Kinesis Agent for Windows sink types that access AWS services, edit the Kinesis Agent for Windows configuration file located at %Program Files%\Amazon\KinesisTap\AWSKinesisTap.exe.config. For instructions, see the proxy section in Configuration Files Reference for AWS SDK for .NET in the AWS SDK for .NET Developer Guide.

Configuring resolving variables in more sink attributes

The following example shows a sink configuration that uses the Region environment variable for the value of the Region attribute key-value pair. For RoleARN, it specifies the EC2 tag key MyRoleARN, which evaluates to the value associated with that key.

"Id": "myCloudWatchLogsSink",
"SinkType": "CloudWatchLogs",
"LogGroup": "EC2Logs",
"LogStream": "logs-{instance_id}"
"Region": "{env:Region}"
"RoleARN": "{ec2tag:MyRoleARN}"

Configuring AWS STS Regional Endpoints When Using RoleARN Property in AWS Sinks

This feature only applies if you are using KinesisTap on Amazon EC2 and using the RoleARN property of AWS sinks to assume an external IAM role to authenticate with the destination AWS services.

By setting UseSTSRegionalEndpoints to true, you can specify that an agent use the regional endpoint (for example, https://sts.us-east-1.amazonaws.com) instead of the global endpoint (for example, https://sts.amazonaws.com). Using a Regional STS endpoint reduces round-trip latency for the operation and limits the impact of failures in the global endpoint service.

Configuring VPC Endpoint for AWS Sinks

You can specify a VPC endpoint in the sink configuration for CloudWatchLogs, CloudWatch, KinesisStreams, and KinesisFirehose sink types. A VPC endpoint enables you to privately connect your VPC to supported AWS services and VPC endpoint services powered by AWS PrivateLink without requiring an internet gateway, NAT device, VPN connection, or AWS Direct Connect connection. Instances in your VPC do not require public IP addresses to communicate with resources in the service. Traffic between your VPC and the other service does not leave the Amazon network. For more information, see VPC endpoints in the Amazon VPC User Guide.

You specify the VPC endpoint using the ServiceURL property as shown in the following example of a CloudWatchLogs sink configuration. Set the value of ServiceURL to the value shown on the VPC endpoint details tab using the Amazon VPC console.

{
    "Id": "myCloudWatchLogsSink",
    "SinkType": "CloudWatchLogs",
    "LogGroup": "EC2Logs",
    "LogStream": "logs-{instance_id}",
    "ServiceURL":"https://vpce-ab1c234de56-ab7cdefg.logs.us-east-1.vpce.amazonaws.com"
}

Configuring An Alternate Means of Proxy

This feature allows you to configure a proxy server in a sink configuration using the proxy support built in to the AWS SDK instead of .NET. Previously, the only way to configure the agent to use a proxy was to use a native feature of .NET, which automatically routed all HTTP/S requests through the proxy defined in the proxy file.

If you are currently using the agent with a proxy server, you do not need to change over to use this method.

You can use the ProxyHost and ProxyPort properties to configure an alternate proxy as shown in the following example.

{
    "Id": "myCloudWatchLogsSink",
    "SinkType": "CloudWatchLogs",
    "LogGroup": "EC2Logs",
    "LogStream": "logs-{instance_id}",
    "Region": "us-west-2",
    "ProxyHost": "myproxy.mydnsdomain.com",
    "ProxyPort": "8080"
}