func NewCloudwatchUploader(adapter *CloudwatchAdapter) *CloudwatchUploader {
	region := adapter.Route.Address
	if (region == "auto") || (region == "") {
		if adapter.Ec2Region == "" {
			log.Println("cloudwatch: ERROR - could not get region from EC2")
		} else {
			region = adapter.Ec2Region
		}
	}
	debugSet := false
	_, debugOption := adapter.Route.Options[`DEBUG`]
	if debugOption || (os.Getenv(`DEBUG`) != "") {
		debugSet = true
		log.Println("cloudwatch: Creating AWS Cloudwatch client for region",
			region)
	}
	uploader := CloudwatchUploader{
		Input:    make(chan CloudwatchBatch),
		tokens:   map[string]string{},
		debugSet: debugSet,
		svc: cloudwatchlogs.New(session.New(),
			&aws.Config{Region: aws.String(region)}),
	}
	go uploader.Start()
	return &uploader
}

// newAWSLogsClient creates the service client for Amazon CloudWatch Logs.
// Customizations to the default client from the SDK include a Docker-specific
// User-Agent string and automatic region detection using the EC2 Instance
// Metadata Service when region is otherwise unspecified.
func newAWSLogsClient(ctx logger.Context) (api, error) {
	config := defaults.DefaultConfig
	if ctx.Config[regionKey] != "" {
		config = defaults.DefaultConfig.Merge(&aws.Config{
			Region: aws.String(ctx.Config[regionKey]),
		})
	}
	if config.Region == nil || *config.Region == "" {
		logrus.Info("Trying to get region from EC2 Metadata")
		ec2MetadataClient := newRegionFinder()
		region, err := ec2MetadataClient.Region()
		if err != nil {
			logrus.WithFields(logrus.Fields{
				"error": err,
			}).Error("Could not get region from EC2 metadata, environment, or log option")
			return nil, errors.New("Cannot determine region for awslogs driver")
		}
		config.Region = &region
	}
	logrus.WithFields(logrus.Fields{
		"region": *config.Region,
	}).Debug("Created awslogs client")
	client := cloudwatchlogs.New(config)

	client.Handlers.Build.PushBackNamed(request.NamedHandler{
		Name: "DockerUserAgentHandler",
		Fn: func(r *request.Request) {
			currentAgent := r.HTTPRequest.Header.Get(userAgentHeader)
			r.HTTPRequest.Header.Set(userAgentHeader,
				fmt.Sprintf("Docker %s (%s) %s",
					dockerversion.Version, runtime.GOOS, currentAgent))
		},
	})
	return client, nil
}

// New creates an awslogs logger using the configuration passed in on the
// context.  Supported context configuration variables are awslogs-region,
// awslogs-group, and awslogs-stream.  When available, configuration is
// also taken from environment variables AWS_REGION, AWS_ACCESS_KEY_ID,
// AWS_SECRET_ACCESS_KEY, the shared credentials file (~/.aws/credentials), and
// the EC2 Instance Metadata Service.
func New(ctx logger.Context) (logger.Logger, error) {
	logGroupName := ctx.Config[logGroupKey]
	logStreamName := ctx.ContainerID
	if ctx.Config[logStreamKey] != "" {
		logStreamName = ctx.Config[logStreamKey]
	}
	config := aws.DefaultConfig
	if ctx.Config[regionKey] != "" {
		config = aws.DefaultConfig.Merge(&aws.Config{
			Region: aws.String(ctx.Config[regionKey]),
		})
	}
	containerStream := &logStream{
		logStreamName: logStreamName,
		logGroupName:  logGroupName,
		client:        cloudwatchlogs.New(config),
		messages:      make(chan *logger.Message, 4096),
	}
	err := containerStream.create()
	if err != nil {
		return nil, err
	}
	go containerStream.collectBatch()

	return containerStream, nil
}

// Logs returns logs.
func (p *Project) Logs(s *session.Session, name string, filter string, duration string) (*logs.Logs, error) {
	fn, err := p.FunctionByName(name)
	if err != nil {
		return nil, err
	}

	fnName, err := p.name(fn)
	if err != nil {
		return nil, err
	}

	l := &logs.Logs{
		Service:       cloudwatchlogs.New(s),
		Log:           log.Log,
		GroupName:     fmt.Sprintf("/aws/lambda/%s", fnName),
		FilterPattern: filter,
	}

	start := time.Now().Add(-time.Minute)
	end := time.Now()

	if duration != "" {
		d, err := time.ParseDuration(duration)
		if err != nil {
			return nil, err
		}
		start = time.Now().Add(-d)
	}

	l.StartTime = start
	l.EndTime = end

	return l, nil
}

func ExampleCloudWatchLogs_PutLogEvents() {
	sess, err := session.NewSession()
	if err != nil {
		fmt.Println("failed to create session,", err)
		return
	}

	svc := cloudwatchlogs.New(sess)

	params := &cloudwatchlogs.PutLogEventsInput{
		LogEvents: []*cloudwatchlogs.InputLogEvent{ // Required
			{ // Required
				Message:   aws.String("EventMessage"), // Required
				Timestamp: aws.Int64(1),               // Required
			},
			// More values...
		},
		LogGroupName:  aws.String("LogGroupName"),  // Required
		LogStreamName: aws.String("LogStreamName"), // Required
		SequenceToken: aws.String("SequenceToken"),
	}
	resp, err := svc.PutLogEvents(params)

	if err != nil {
		// Print the error, cast err to awserr.Error to get the Code and
		// Message from an error.
		fmt.Println(err.Error())
		return
	}

	// Pretty-print the response data.
	fmt.Println(resp)
}

// Run command.
func run(c *cobra.Command, args []string) error {
	stats.Track("Logs", map[string]interface{}{
		"has_filter": filter != "",
		"since":      duration,
		"follow":     follow,
	})

	if err := root.Project.LoadFunctions(args...); err != nil {
		return err
	}

	config := logs.Config{
		Service:       cloudwatchlogs.New(root.Session),
		StartTime:     time.Now().Add(-duration).UTC(),
		PollInterval:  5 * time.Second,
		Follow:        follow,
		FilterPattern: filter,
	}

	l := &logs.Logs{
		Config: config,
	}

	for _, fn := range root.Project.Functions {
		l.GroupNames = append(l.GroupNames, fn.GroupName())
	}

	for event := range l.Start() {
		fmt.Printf("\033[34m%s\033[0m %s", event.GroupName, event.Message)
	}

	return l.Err()
}

func ExampleCloudWatchLogs_GetLogEvents() {
	sess, err := session.NewSession()
	if err != nil {
		fmt.Println("failed to create session,", err)
		return
	}

	svc := cloudwatchlogs.New(sess)

	params := &cloudwatchlogs.GetLogEventsInput{
		LogGroupName:  aws.String("LogGroupName"),  // Required
		LogStreamName: aws.String("LogStreamName"), // Required
		EndTime:       aws.Int64(1),
		Limit:         aws.Int64(1),
		NextToken:     aws.String("NextToken"),
		StartFromHead: aws.Bool(true),
		StartTime:     aws.Int64(1),
	}
	resp, err := svc.GetLogEvents(params)

	if err != nil {
		// Print the error, cast err to awserr.Error to get the Code and
		// Message from an error.
		fmt.Println(err.Error())
		return
	}

	// Pretty-print the response data.
	fmt.Println(resp)
}

func ExampleCloudWatchLogs_DescribeSubscriptionFilters() {
	sess, err := session.NewSession()
	if err != nil {
		fmt.Println("failed to create session,", err)
		return
	}

	svc := cloudwatchlogs.New(sess)

	params := &cloudwatchlogs.DescribeSubscriptionFiltersInput{
		LogGroupName:     aws.String("LogGroupName"), // Required
		FilterNamePrefix: aws.String("FilterName"),
		Limit:            aws.Int64(1),
		NextToken:        aws.String("NextToken"),
	}
	resp, err := svc.DescribeSubscriptionFilters(params)

	if err != nil {
		// Print the error, cast err to awserr.Error to get the Code and
		// Message from an error.
		fmt.Println(err.Error())
		return
	}

	// Pretty-print the response data.
	fmt.Println(resp)
}

func ExampleCloudWatchLogs_CreateExportTask() {
	sess, err := session.NewSession()
	if err != nil {
		fmt.Println("failed to create session,", err)
		return
	}

	svc := cloudwatchlogs.New(sess)

	params := &cloudwatchlogs.CreateExportTaskInput{
		Destination:         aws.String("ExportDestinationBucket"), // Required
		From:                aws.Int64(1),                          // Required
		LogGroupName:        aws.String("LogGroupName"),            // Required
		To:                  aws.Int64(1),                          // Required
		DestinationPrefix:   aws.String("ExportDestinationPrefix"),
		LogStreamNamePrefix: aws.String("LogStreamName"),
		TaskName:            aws.String("ExportTaskName"),
	}
	resp, err := svc.CreateExportTask(params)

	if err != nil {
		// Print the error, cast err to awserr.Error to get the Code and
		// Message from an error.
		fmt.Println(err.Error())
		return
	}

	// Pretty-print the response data.
	fmt.Println(resp)
}

func ExampleCloudWatchLogs_DescribeExportTasks() {
	sess, err := session.NewSession()
	if err != nil {
		fmt.Println("failed to create session,", err)
		return
	}

	svc := cloudwatchlogs.New(sess)

	params := &cloudwatchlogs.DescribeExportTasksInput{
		Limit:      aws.Int64(1),
		NextToken:  aws.String("NextToken"),
		StatusCode: aws.String("ExportTaskStatusCode"),
		TaskId:     aws.String("ExportTaskId"),
	}
	resp, err := svc.DescribeExportTasks(params)

	if err != nil {
		// Print the error, cast err to awserr.Error to get the Code and
		// Message from an error.
		fmt.Println(err.Error())
		return
	}

	// Pretty-print the response data.
	fmt.Println(resp)
}

func ExampleCloudWatchLogs_PutSubscriptionFilter() {
	svc := cloudwatchlogs.New(nil)

	params := &cloudwatchlogs.PutSubscriptionFilterInput{
		DestinationARN: aws.String("DestinationArn"), // Required
		FilterName:     aws.String("FilterName"),     // Required
		FilterPattern:  aws.String("FilterPattern"),  // Required
		LogGroupName:   aws.String("LogGroupName"),   // Required
		RoleARN:        aws.String("RoleArn"),        // Required
	}
	resp, err := svc.PutSubscriptionFilter(params)

	if err != nil {
		if awsErr, ok := err.(awserr.Error); ok {
			// Generic AWS Error with Code, Message, and original error (if any)
			fmt.Println(awsErr.Code(), awsErr.Message(), awsErr.OrigErr())
			if reqErr, ok := err.(awserr.RequestFailure); ok {
				// A service error occurred
				fmt.Println(reqErr.Code(), reqErr.Message(), reqErr.StatusCode(), reqErr.RequestID())
			}
		} else {
			// This case should never be hit, the SDK should always return an
			// error which satisfies the awserr.Error interface.
			fmt.Println(err.Error())
		}
	}

	// Pretty-print the response data.
	fmt.Println(awsutil.StringValue(resp))
}

func ExampleCloudWatchLogs_GetLogEvents() {
	svc := cloudwatchlogs.New(nil)

	params := &cloudwatchlogs.GetLogEventsInput{
		LogGroupName:  aws.String("LogGroupName"),  // Required
		LogStreamName: aws.String("LogStreamName"), // Required
		EndTime:       aws.Long(1),
		Limit:         aws.Long(1),
		NextToken:     aws.String("NextToken"),
		StartFromHead: aws.Boolean(true),
		StartTime:     aws.Long(1),
	}
	resp, err := svc.GetLogEvents(params)

	if err != nil {
		if awsErr, ok := err.(awserr.Error); ok {
			// Generic AWS Error with Code, Message, and original error (if any)
			fmt.Println(awsErr.Code(), awsErr.Message(), awsErr.OrigErr())
			if reqErr, ok := err.(awserr.RequestFailure); ok {
				// A service error occurred
				fmt.Println(reqErr.Code(), reqErr.Message(), reqErr.StatusCode(), reqErr.RequestID())
			}
		} else {
			// This case should never be hit, the SDK should always return an
			// error which satisfies the awserr.Error interface.
			fmt.Println(err.Error())
		}
	}

	// Pretty-print the response data.
	fmt.Println(awsutil.StringValue(resp))
}

func ExampleCloudWatchLogs_PutDestinationPolicy() {
	sess, err := session.NewSession()
	if err != nil {
		fmt.Println("failed to create session,", err)
		return
	}

	svc := cloudwatchlogs.New(sess)

	params := &cloudwatchlogs.PutDestinationPolicyInput{
		AccessPolicy:    aws.String("AccessPolicy"),    // Required
		DestinationName: aws.String("DestinationName"), // Required
	}
	resp, err := svc.PutDestinationPolicy(params)

	if err != nil {
		// Print the error, cast err to awserr.Error to get the Code and
		// Message from an error.
		fmt.Println(err.Error())
		return
	}

	// Pretty-print the response data.
	fmt.Println(resp)
}

func ExampleCloudWatchLogs_PutMetricFilter() {
	sess, err := session.NewSession()
	if err != nil {
		fmt.Println("failed to create session,", err)
		return
	}

	svc := cloudwatchlogs.New(sess)

	params := &cloudwatchlogs.PutMetricFilterInput{
		FilterName:    aws.String("FilterName"),    // Required
		FilterPattern: aws.String("FilterPattern"), // Required
		LogGroupName:  aws.String("LogGroupName"),  // Required
		MetricTransformations: []*cloudwatchlogs.MetricTransformation{ // Required
			{ // Required
				MetricName:      aws.String("MetricName"),      // Required
				MetricNamespace: aws.String("MetricNamespace"), // Required
				MetricValue:     aws.String("MetricValue"),     // Required
				DefaultValue:    aws.Float64(1.0),
			},
			// More values...
		},
	}
	resp, err := svc.PutMetricFilter(params)

	if err != nil {
		// Print the error, cast err to awserr.Error to get the Code and
		// Message from an error.
		fmt.Println(err.Error())
		return
	}

	// Pretty-print the response data.
	fmt.Println(resp)
}

func main() {
	g := cloudwatch.NewGroup("test", cloudwatchlogs.New(defaults.DefaultConfig))

	stream := uuid.New()

	w, err := g.Create(stream)
	if err != nil {
		log.Fatal(err)
	}

	r, err := g.Open(stream)
	if err != nil {
		log.Fatal(err)
	}

	go func() {
		var i int
		for {
			i++
			<-time.After(time.Second / 30)
			_, err := fmt.Fprintf(w, "Line %d\n", i)
			if err != nil {
				log.Println(err)
			}
		}
	}()

	io.Copy(os.Stdout, r)
}

func ExampleCloudWatchLogs_TestMetricFilter() {
	sess, err := session.NewSession()
	if err != nil {
		fmt.Println("failed to create session,", err)
		return
	}

	svc := cloudwatchlogs.New(sess)

	params := &cloudwatchlogs.TestMetricFilterInput{
		FilterPattern: aws.String("FilterPattern"), // Required
		LogEventMessages: []*string{ // Required
			aws.String("EventMessage"), // Required
			// More values...
		},
	}
	resp, err := svc.TestMetricFilter(params)

	if err != nil {
		// Print the error, cast err to awserr.Error to get the Code and
		// Message from an error.
		fmt.Println(err.Error())
		return
	}

	// Pretty-print the response data.
	fmt.Println(resp)
}

func ExampleCloudWatchLogs_DeleteLogGroup() {
	svc := cloudwatchlogs.New(nil)

	params := &cloudwatchlogs.DeleteLogGroupInput{
		LogGroupName: aws.String("LogGroupName"), // Required
	}
	resp, err := svc.DeleteLogGroup(params)

	if err != nil {
		if awsErr, ok := err.(awserr.Error); ok {
			// Generic AWS error with Code, Message, and original error (if any)
			fmt.Println(awsErr.Code(), awsErr.Message(), awsErr.OrigErr())
			if reqErr, ok := err.(awserr.RequestFailure); ok {
				// A service error occurred
				fmt.Println(reqErr.Code(), reqErr.Message(), reqErr.StatusCode(), reqErr.RequestID())
			}
		} else {
			// This case should never be hit, the SDK should always return an
			// error which satisfies the awserr.Error interface.
			fmt.Println(err.Error())
		}
	}

	// Pretty-print the response data.
	fmt.Println(awsutil.Prettify(resp))
}

func ExampleCloudWatchLogs_PutLogEvents() {
	svc := cloudwatchlogs.New(nil)

	params := &cloudwatchlogs.PutLogEventsInput{
		LogEvents: []*cloudwatchlogs.InputLogEvent{ // Required
			{ // Required
				Message:   aws.String("EventMessage"), // Required
				Timestamp: aws.Int64(1),               // Required
			},
			// More values...
		},
		LogGroupName:  aws.String("LogGroupName"),  // Required
		LogStreamName: aws.String("LogStreamName"), // Required
		SequenceToken: aws.String("SequenceToken"),
	}
	resp, err := svc.PutLogEvents(params)

	if err != nil {
		if awsErr, ok := err.(awserr.Error); ok {
			// Generic AWS error with Code, Message, and original error (if any)
			fmt.Println(awsErr.Code(), awsErr.Message(), awsErr.OrigErr())
			if reqErr, ok := err.(awserr.RequestFailure); ok {
				// A service error occurred
				fmt.Println(reqErr.Code(), reqErr.Message(), reqErr.StatusCode(), reqErr.RequestID())
			}
		} else {
			// This case should never be hit, the SDK should always return an
			// error which satisfies the awserr.Error interface.
			fmt.Println(err.Error())
		}
	}

	// Pretty-print the response data.
	fmt.Println(awsutil.Prettify(resp))
}

func ExampleCloudWatchLogs_TestMetricFilter() {
	svc := cloudwatchlogs.New(nil)

	params := &cloudwatchlogs.TestMetricFilterInput{
		FilterPattern: aws.String("FilterPattern"), // Required
		LogEventMessages: []*string{ // Required
			aws.String("EventMessage"), // Required
			// More values...
		},
	}
	resp, err := svc.TestMetricFilter(params)

	if err != nil {
		if awsErr, ok := err.(awserr.Error); ok {
			// Generic AWS error with Code, Message, and original error (if any)
			fmt.Println(awsErr.Code(), awsErr.Message(), awsErr.OrigErr())
			if reqErr, ok := err.(awserr.RequestFailure); ok {
				// A service error occurred
				fmt.Println(reqErr.Code(), reqErr.Message(), reqErr.StatusCode(), reqErr.RequestID())
			}
		} else {
			// This case should never be hit, the SDK should always return an
			// error which satisfies the awserr.Error interface.
			fmt.Println(err.Error())
		}
	}

	// Pretty-print the response data.
	fmt.Println(awsutil.Prettify(resp))
}

func ExampleCloudWatchLogs_CreateLogStream() {
	sess, err := session.NewSession()
	if err != nil {
		fmt.Println("failed to create session,", err)
		return
	}

	svc := cloudwatchlogs.New(sess)

	params := &cloudwatchlogs.CreateLogStreamInput{
		LogGroupName:  aws.String("LogGroupName"),  // Required
		LogStreamName: aws.String("LogStreamName"), // Required
	}
	resp, err := svc.CreateLogStream(params)

	if err != nil {
		// Print the error, cast err to awserr.Error to get the Code and
		// Message from an error.
		fmt.Println(err.Error())
		return
	}

	// Pretty-print the response data.
	fmt.Println(resp)
}