func (prod *AwsCloudwatchLogs) upload(messages []*core.Message) {
	messages = prod.processBatch(messages)
	if len(messages) == 0 {
		return
	}
	logevents := make([]*cloudwatchlogs.InputLogEvent, 0, len(messages))
	for _, msg := range messages {
		logevents = append(logevents, &cloudwatchlogs.InputLogEvent{
			Message:   aws.String(msg.String()),
			Timestamp: aws.Int64(msg.GetCreationTime().Unix() * 1000),
		})
	}
	params := &cloudwatchlogs.PutLogEventsInput{
		LogEvents:     logevents,
		LogGroupName:  aws.String(prod.group),
		LogStreamName: aws.String(prod.stream),
		SequenceToken: prod.token,
	}
	prod.handleResult(prod.service.PutLogEvents(params))
}
Put log events and update sequence token.
Possible errors http://docs.aws.amazon.com/AmazonCloudWatchLogs/latest/APIReference/API_PutLogEvents.html
package producer
import (
	"sort"
	"sync"
	"time"
	"gollum/core"
	"gollum/core/components"
	"github.com/aws/aws-sdk-go/aws"
	"github.com/aws/aws-sdk-go/aws/awserr"
	"github.com/aws/aws-sdk-go/aws/endpoints"
	"github.com/aws/aws-sdk-go/service/cloudwatchlogs"
)
const (
	maxBatchEvents = 10000
	maxBatchSize = 1048576
	maxBatchTimeSpan = 24 * time.Hour
	eventSizeOverhead = 26
)
type AwsCloudwatchLogs struct {
	AwsMultiClient       components.AwsMultiClient `gollumdoc:"embed_type"`
	core.BatchedProducer `gollumdoc:"embed_type"`
	stream               string `config:"LogStream" default:""`
	group                string `config:"LogGroup" default:""`
	token                *string
	service              *cloudwatchlogs.CloudWatchLogs
}
func init() {
	core.TypeRegistry.Register(AwsCloudwatchLogs{})
}
func (prod *AwsCloudwatchLogs) Configure(conf core.PluginConfigReader) {
	if prod.stream == "" {
		prod.Logger.Error("LogStream can not be empty")
	}
	if prod.group == "" {
		prod.Logger.Error("LogGroup can not be empty")
	}
	if conf.GetInt("Batch/MaxCount", maxBatchEvents) > maxBatchEvents {
		conf.Errors.Pushf("Batch/MaxCount must be below %d", maxBatchEvents)
	}
}
func (m byTimestamp) Len() int {
	return len(m)
}
func (m byTimestamp) Swap(i, j int) {
	m[i], m[j] = m[j], m[i]
}
func (m byTimestamp) Less(i, j int) bool {
	return m[i].GetCreationTime().Unix() < m[j].GetCreationTime().Unix()
}
type byTimestamp []*core.Message
func (prod *AwsCloudwatchLogs) numEvents(messages []*core.Message, checkFn ...indexNumFn) int {
	index := maxBatchEvents
	for _, fn := range checkFn {
		result := fn(messages)
		if result < index {
			index = result
		}
	}
	return index
}
type indexNumFn func(messages []*core.Message) int
func (prod *AwsCloudwatchLogs) sizeIndex(messages []*core.Message) int {
	size, index := 0, 0
	for i, message := range messages {
		size += len(message.String()) + eventSizeOverhead
		if size > maxBatchSize {
			break
		}
		index = i + 1
	}
	return index
}
func (prod *AwsCloudwatchLogs) timeIndex(messages []*core.Message) (index int) {
	if len(messages) == 0 {
		return 0
	}
	firstTimestamp := messages[0].GetCreationTime().Unix()
	for i, message := range messages {
		if (message.GetCreationTime().Unix() - firstTimestamp) > int64(maxBatchTimeSpan) {
			break
		}
		index = i + 1
	}
	return index
}
func (prod *AwsCloudwatchLogs) processBatch(messages []*core.Message) []*core.Message {
	sort.Sort(byTimestamp(messages))
	return messages[:prod.numEvents(messages, prod.sizeIndex, prod.timeIndex)]
}