func (s *S3Sink) upload() {
	now := time.Now()
	key := s.getNewKey(now)
	_, err := s.uploader.Upload(&s3manager.UploadInput{
		Bucket: aws.String(s.bucket),
		Key:    aws.String(key),
		Body:   s.bodyBuf,
	})
	if err != nil {
		glog.Errorf("Error uploading %s to s3, %v", key, err)
	}
	glog.Infof("Uploaded at %s", key)
	s.lastUploadTimestamp = now.UnixNano()
	s.bodyBuf.Truncate(0)
}
upload uploads the events stored in buffer to s3 in the specified key
and clears the buffer
package sinks
import (
	"bytes"
	"errors"
	"fmt"
	"time"
	"k8s.io/api/core/v1"
	"github.com/aws/aws-sdk-go/aws"
	"github.com/aws/aws-sdk-go/aws/credentials"
	"github.com/aws/aws-sdk-go/aws/session"
	"github.com/aws/aws-sdk-go/service/s3/s3manager"
	"github.com/eapache/channels"
	"github.com/golang/glog"
)
type S3Sink struct {
	uploader *s3manager.Uploader
	bucket string
	bucketDir string
	outputFormat string
	lastUploadTimestamp int64
	uploadInterval time.Duration
	eventCh channels.Channel
	bodyBuf *bytes.Buffer
}
func NewS3Sink(awsAccessKeyID string, s3SinkSecretAccessKey string, s3SinkRegion string, s3SinkBucket string, s3SinkBucketDir string, s3SinkUploadInterval int, overflow bool, bufferSize int, outputFormat string) (*S3Sink, error) {
	awsConfig := &aws.Config{
		Region:      aws.String(s3SinkRegion),
		Credentials: credentials.NewStaticCredentials(awsAccessKeyID, s3SinkSecretAccessKey, ""),
	}
	awsConfig = awsConfig.WithCredentialsChainVerboseErrors(true)
	sess, err := session.NewSession(awsConfig)
	if err != nil {
		return nil, err
	}
	uploader := s3manager.NewUploader(sess)
	s := &S3Sink{
		uploader:       uploader,
		bucket:         s3SinkBucket,
		bucketDir:      s3SinkBucketDir,
		uploadInterval: time.Second * time.Duration(s3SinkUploadInterval),
		outputFormat:   outputFormat,
		bodyBuf:        bytes.NewBuffer(make([]byte, 0, 4096)),
	}
	if overflow {
		s.eventCh = channels.NewOverflowingChannel(channels.BufferCap(bufferSize))
	} else {
		s.eventCh = channels.NewNativeChannel(channels.BufferCap(bufferSize))
	}
	return s, nil
}
func (s *S3Sink) UpdateEvents(eNew *v1.Event, eOld *v1.Event) {
	s.eventCh.In() <- NewEventData(eNew, eOld)
}
func (s *S3Sink) Run(stopCh <-chan bool) {
loop:
	for {
		select {
		case e := <-s.eventCh.Out():
			var evt EventData
			var ok bool
			if evt, ok = e.(EventData); !ok {
				glog.Warningf("Invalid type sent through event channel: %T", e)
				continue loop
			}
			arr := []EventData{evt}
			numEvents := s.eventCh.Len()
			for i := 0; i < numEvents; i++ {
				e := <-s.eventCh.Out()
				if evt, ok = e.(EventData); ok {
					arr = append(arr, evt)
				} else {
					glog.Warningf("Invalid type sent through event channel: %T", e)
				}
			}
			s.drainEvents(arr)
		case <-stopCh:
			break loop
		}
	}
}
func (s *S3Sink) drainEvents(events []EventData) {
	var written int64
	for _, evt := range events {
		switch s.outputFormat {
		case "rfc5424":
			w, err := evt.WriteRFC5424(s.bodyBuf)
			written += w
			if err != nil {
				glog.Warningf("Could not write to event request body (wrote %v) bytes: %v", written, err)
				return
			}
		case "flatjson":
			w, err := evt.WriteFlattenedJSON(s.bodyBuf)
			written += w
			if err != nil {
				glog.Warningf("Could not write to event request body (wrote %v) bytes: %v", written, err)
				return
			}
		default:
			err := errors.New("Invalid Sink Output Format specified")
			panic(err.Error())
		}
		s.bodyBuf.Write([]byte{'\n'})
		written++
	}
	if s.canUpload() == false {
		return
	}
	s.upload()
}
func (s *S3Sink) canUpload() bool {
	now := time.Now().UnixNano()
	if (s.lastUploadTimestamp + s.uploadInterval.Nanoseconds()) < now {
		return true
	}
	return false
}
func (s *S3Sink) getNewKey(t time.Time) string {
	return fmt.Sprintf("%s/%d/%d/%d/%d.txt", s.bucketDir, t.Year(), t.Month(), t.Day(), t.UnixNano())
}