func (c *CLI) getListRDSLogFiles() ([]LogFile, error) {
	var output *rds.DescribeDBLogFilesOutput
	var err error
	var logFiles []LogFile
	for {
		if output == nil {
			output, err = c.RDS.DescribeDBLogFiles(&rds.DescribeDBLogFilesInput{
				DBInstanceIdentifier: &c.Options.InstanceIdentifier,
			})
			logFiles = make([]LogFile, 0, len(output.DescribeDBLogFiles))
		} else {
			output, err = c.RDS.DescribeDBLogFiles(&rds.DescribeDBLogFilesInput{
				DBInstanceIdentifier: &c.Options.InstanceIdentifier,
				Marker:               output.Marker,
			})
		}
		if err != nil {
			return nil, err
		}
		for _, lf := range output.DescribeDBLogFiles {
			logFiles = append(logFiles, LogFile{
				LastWritten:     *lf.LastWritten,
				LastWrittenTime: time.Unix(*lf.LastWritten/1000, 0),
				LogFileName:     *lf.LogFileName,
				Size:            *lf.Size,
			})
		}
		if output.Marker == nil {
			break
		}
	}
	return logFiles, nil
}
Gets a list of all available RDS log files for an instance.
package cli
import (
	"errors"
	"fmt"
	"io"
	"os"
	"path"
	"sort"
	"strconv"
	"strings"
	"time"
	"github.com/aws/aws-sdk-go/aws"
	"github.com/aws/aws-sdk-go/service/rds"
	"github.com/honeycombio/honeytail/parsers"
	"github.com/honeycombio/honeytail/parsers/csv"
	"github.com/honeycombio/honeytail/parsers/mysql"
	"github.com/honeycombio/honeytail/parsers/postgresql"
	"github.com/honeycombio/rdslogs/publisher"
	"github.com/sirupsen/logrus"
)
const rdsPostgresLinePrefix = "%t:%r:%u@%d:[%p]:"
const DBTypePostgreSQL = "postgresql"
const DBTypeMySQL = "mysql"
const LogTypeQuery = "query"
const LogTypeAudit = "audit"
type Options struct {
	Region             string            `long:"region" description:"AWS region to use" default:"us-east-1"`
	InstanceIdentifier string            `short:"i" long:"identifier" description:"RDS instance identifier"`
	DBType             string            `long:"dbtype" description:"RDS database type. Accepted values are mysql and postgresql." default:"mysql"`
	LogType            string            `long:"log_type" description:"Log file type. Accepted values are query and audit. Audit is currently only supported for mysql." default:"query"`
	LogFile            string            `short:"f" long:"log_file" description:"RDS log file to retrieve"`
	Download           bool              `short:"d" long:"download" description:"Download old logs instead of tailing the current log"`
	DownloadDir        string            `long:"download_dir" description:"directory in to which log files are downloaded" default:"./"`
	NumLines           int64             `long:"num_lines" description:"number of lines to request at a time from AWS. Larger number will be more efficient, smaller number will allow for longer lines" default:"10000"`
	BackoffTimer       int64             `long:"backoff_timer" description:"how many seconds to pause when rate limited by AWS." default:"5"`
	Output             string            `short:"o" long:"output" description:"output for the logs: stdout or honeycomb" default:"stdout"`
	WriteKey           string            `long:"writekey" description:"Team write key, when output is honeycomb"`
	Dataset            string            `long:"dataset" description:"Name of the dataset, when output is honeycomb"`
	APIHost            string            `long:"api_host" description:"Hostname for the Honeycomb API server" default:"https://api.honeycomb.io/"`
	ScrubQuery         bool              `long:"scrub_query" description:"Replaces the query field with a one-way hash of the contents"`
	SampleRate         int               `long:"sample_rate" description:"Only send 1 / N log lines" default:"1"`
	AddFields          map[string]string `short:"a" long:"add_field" description:"Extra fields to send in request, in the style of \"field:value\""`
	NumParsers         int               `long:"num_parsers" default:"4" description:"Number of parsers to spin up. Currently only supported for the mysql parser."`
	Version            bool   `short:"v" long:"version" description:"Output the current version and exit"`
	ConfigFile         string `short:"c" long:"config" description:"config file" no-ini:"true"`
	WriteDefaultConfig bool   `long:"write_default_config" description:"Write a default config file to STDOUT" no-ini:"true"`
	Debug              bool   `long:"debug" description:"turn on debugging output"`
}
var Usage = `rdslogs --identifier my-rds-instance
rdslogs streams a log file from Amazon RDS and prints it to STDOUT or sends it
up to Honeycomb.io.
AWS credentials are required and can be provided via IAM roles, AWS shared
config (~/.aws/config), AWS shared credentials (~/.aws/credentials), or
the environment variables AWS_ACCESS_KEY_ID and AWS_SECRET_ACCESS_KEY.
Passing --download triggers Download Mode, in which rdslogs will download the
specified logs to the directory specified by --download_dir. Logs are specified
via the --log_file flag, which names an active log file as well as the past 24
hours of rotated logs. (For example, specifying --log_file=foo.log will download
foo.log as well as foo.log.0, foo.log.2, ... foo.log.23.)
When --output is set to "honeycomb", the --writekey and --dataset flags are
required. Instead of being printed to STDOUT, database events from the log will
be transmitted to Honeycomb. --scrub_query and --sample_rate also only apply to
honeycomb output.
`
type CLI struct {
	Options *Options
	RDS *rds.RDS
	Abort chan bool
	output publisher.Publisher
	fakeNower Nower
}
func (c *CLI) Stream() error {
	latestFile, err := c.GetLatestLogFile()
	if err != nil {
		return err
	}
	if c.Options.Output == "stdout" {
		c.output = &publisher.STDOUTPublisher{}
	} else {
		var parser parsers.Parser
		if c.Options.DBType == DBTypeMySQL && c.Options.LogType == LogTypeQuery {
			parser = &mysql.Parser{}
			parser.Init(&mysql.Options{NumParsers: c.Options.NumParsers})
		} else if c.Options.DBType == DBTypeMySQL && c.Options.LogType == LogTypeAudit {
			parser = &csv.Parser{}
			parser.Init(&csv.Options{
				Fields:          "time,hostname,user,source_addr,connection_id,query_id,event_type,database,query,error_code",
				NumParsers:      c.Options.NumParsers,
				TimeFieldName:   "time",
				TimeFieldFormat: "20060102 15:04:05",
			})
		} else if c.Options.DBType == DBTypePostgreSQL {
			parser = &postgresql.Parser{}
			parser.Init(&postgresql.Options{LogLinePrefix: rdsPostgresLinePrefix})
		} else {
			return fmt.Errorf(
				"Unsupported (dbtype, log_type) pair (`%s`,`%s`)",
				c.Options.DBType, c.Options.LogType)
		}
		pub := &publisher.HoneycombPublisher{
			Writekey:   c.Options.WriteKey,
			Dataset:    c.Options.Dataset,
			APIHost:    c.Options.APIHost,
			ScrubQuery: c.Options.ScrubQuery,
			SampleRate: c.Options.SampleRate,
			AddFields:  c.Options.AddFields,
			Parser:     parser,
		}
		defer pub.Close()
		c.output = pub
	}
	sPos := StreamPos{
		logFile: LogFile{LogFileName: latestFile.LogFileName},
	}
	if c.Options.DBType == DBTypeMySQL && c.Options.LogType == LogTypeAudit {
		sPos.logFile.LogFileName = c.Options.LogFile
	}
	for {
		select {
		case <-c.Abort:
			return fmt.Errorf("signal triggered exit")
		default:
		}
		resp, err := c.getRecentEntries(sPos)
		if err != nil {
			if strings.HasPrefix(err.Error(), "Throttling: Rate exceeded") {
				logrus.Infof("AWS Rate limit hit; sleeping for %d seconds.\n", c.Options.BackoffTimer)
				c.waitFor(time.Duration(c.Options.BackoffTimer) * time.Second)
				continue
			}
			if strings.HasPrefix(err.Error(), "InvalidParameterValue: This file contains binary data") {
				logrus.Infof("binary data at marker %s, skipping 1000 in marker position\n", sPos.marker)
				newMarker, err := sPos.Add(1000)
				if err != nil {
					return err
				}
				sPos.marker = newMarker
				continue
			}
			if strings.HasPrefix(err.Error(), "DBLogFileNotFoundFault") {
				logrus.WithError(err).
					Warn("log does not appear to exist (rotation ongoing?) - waiting and retrying")
				c.waitFor(time.Second * 5)
				continue
			}
			return err
		}
		if resp.LogFileData != nil {
			c.output.Write(*resp.LogFileData)
		}
		if c.Options.DBType == DBTypeMySQL && c.Options.LogType == LogTypeAudit {
			if sPos.marker == "0" && (resp.Marker != nil && *resp.Marker == "") {
				c.waitFor(time.Second * 5)
				continue
			}
			if (resp.Marker != nil && resp.LogFileData != nil && sPos.marker == *resp.Marker) ||
				!*resp.AdditionalDataPending && resp.LogFileData == nil {
				newestFile, err := c.GetLatestLogFile()
				if err != nil {
					return err
				}
				if newestFile.LogFileName != sPos.logFile.LogFileName {
					logrus.WithFields(logrus.Fields{
						"expectedFile": sPos.logFile.LogFileName,
						"newestFile":   newestFile.LogFileName,
					}).Info("newest file is a rotated file, we appear to be mid-rotation")
					c.waitFor(time.Second * 5)
					continue
				}
				splitMarker := strings.Split(sPos.marker, ":")
				if len(splitMarker) != 2 {
					logrus.WithField("marker", sPos.marker).
						Warn("marker didn't split into two pieces across a colon")
					continue
				}
				offset, _ := strconv.Atoi(splitMarker[1])
				if int64(offset) > newestFile.Size {
					logrus.WithFields(logrus.Fields{
						"currentOffset": offset,
						"newFileSize":   newestFile.Size,
					}).Info("last marker offset exceeds newest file size, resetting marker to 0")
					sPos.marker = "0"
					continue
				}
			}
		}
		if !*resp.AdditionalDataPending || (resp.Marker != nil && *resp.Marker == "0") {
			if c.Options.DBType == DBTypePostgreSQL {
				newestFile, err := c.GetLatestLogFile()
				if err != nil {
					return err
				}
				if newestFile.LogFileName != sPos.logFile.LogFileName {
					logrus.WithFields(logrus.Fields{
						"oldFile": sPos.logFile.LogFileName,
						"newFile": newestFile.LogFileName}).Info("Found newer file")
					sPos = StreamPos{logFile: LogFile{LogFileName: newestFile.LogFileName}}
					continue
				}
			}
			c.waitFor(5 * time.Second)
		}
		newMarker := c.getNextMarker(sPos, resp)
		logrus.WithFields(logrus.Fields{
			"prevMarker": sPos.marker,
			"newMarker":  newMarker,
			"file":       sPos.logFile.LogFileName}).Info("Got new marker")
		sPos.marker = newMarker
	}
}
func (c *CLI) getNextMarker(sPos StreamPos, resp *rds.DownloadDBLogFilePortionOutput) string {
	if resp == nil {
		logrus.Warn("resp was nil, returning previous marker")
		return sPos.marker
	}
	if resp.Marker == nil {
		logrus.Warn("resp marker is nil, returning previous marker")
		return sPos.marker
	}
	if *resp.Marker != "0" {
		return *resp.Marker
	}
	if resp.LogFileData != nil && len(*resp.LogFileData) != 0 {
		newMarkerStr, err := sPos.Add(len(*resp.LogFileData))
		if err != nil {
			logrus.WithError(err).
				Warn("failed to get next marker. Reverting to no marker.")
			return "0"
		}
		return newMarkerStr
	}
	var now time.Time
	if c.fakeNower != nil {
		now = c.fakeNower.Now().UTC()
	} else {
		now = time.Now().UTC()
	}
	curMin, _ := strconv.Atoi(now.Format("04"))
	if curMin > 5 {
		logrus.WithField("newMarker", *resp.Marker).
			Infof("no log data received but it's %d minutes (> 5) past "+
				"the hour, returning resp marker", curMin)
		return *resp.Marker
	}
	logrus.WithField("prevMarker", sPos.marker).
		Infof("no log data received but it's %d minutes (< 5) past "+
			"the hour, returning previous marker", curMin)
	return sPos.marker
}
type StreamPos struct {
	logFile LogFile
	marker  string
}
func (s *StreamPos) Add(dataLen int) (string, error) {
	splitMarker := strings.Split(s.marker, ":")
	if len(splitMarker) != 2 {
		return "", fmt.Errorf("marker didn't split into two pieces across a colon")
	}
	mHour, _ := strconv.Atoi(splitMarker[0])
	mOffset, _ := strconv.Atoi(splitMarker[1])
	mOffset += dataLen
	return fmt.Sprintf("%d:%d", mHour, mOffset), nil
}
func (c *CLI) getRecentEntries(sPos StreamPos) (*rds.DownloadDBLogFilePortionOutput, error) {
	params := &rds.DownloadDBLogFilePortionInput{
		DBInstanceIdentifier: aws.String(c.Options.InstanceIdentifier),
		LogFileName:          aws.String(sPos.logFile.LogFileName),
		NumberOfLines:        aws.Int64(c.Options.NumLines),
	}
	if sPos.marker != "" {
		params.Marker = &sPos.marker
	} else {
		params.NumberOfLines = aws.Int64(1)
	}
	return c.RDS.DownloadDBLogFilePortion(params)
}
func (c *CLI) Download() error {
	logFiles, err := c.GetLogFiles()
	if err != nil {
		return err
	}
	logFiles, err = c.DownloadLogFiles(logFiles)
	if err != nil {
		fmt.Println("Error downloading log files:")
		return err
	}
	return nil
}
type LogFile struct {
	Size            int64  
	LogFileName     string
	LastWritten     int64  
	LastWrittenTime time.Time
	Path            string
}
func (l *LogFile) String() string {
	return fmt.Sprintf("%-35s (date: %s, size: %d)", l.LogFileName, l.LastWrittenTime, l.Size)
}
func (c *CLI) DownloadLogFiles(logFiles []LogFile) ([]LogFile, error) {
	logrus.Infof("Downloading log files to %s\n", c.Options.DownloadDir)
	downloadedLogFiles := make([]LogFile, 0, len(logFiles))
	for i := range logFiles {
		logFile, err := c.downloadFile(logFiles[i])
		if err != nil {
			return nil, err
		}
		downloadedLogFiles = append(downloadedLogFiles, logFile)
	}
	return downloadedLogFiles, nil
}
func (c *CLI) downloadFile(logFile LogFile) (LogFile, error) {
	logFile.Path = path.Join(c.Options.DownloadDir, path.Base(logFile.LogFileName))
	fmt.Printf("Downloading %s to %s ... ", logFile.LogFileName, logFile.Path)
	defer fmt.Printf("done\n")
	if err := os.MkdirAll(path.Dir(logFile.Path), os.ModePerm); err != nil {
		return logFile, err
	}
	outfile, err := os.Create(logFile.Path)
	if err != nil {
		return logFile, err
	}
	defer outfile.Close()
	resp := &rds.DownloadDBLogFilePortionOutput{
		AdditionalDataPending: aws.Bool(true),
		Marker:                aws.String("0"),
	}
	params := &rds.DownloadDBLogFilePortionInput{
		DBInstanceIdentifier: aws.String(c.Options.InstanceIdentifier),
		LogFileName:          aws.String(logFile.LogFileName),
	}
	for aws.BoolValue(resp.AdditionalDataPending) {
		select {
		case <-c.Abort:
			return logFile, fmt.Errorf("signal triggered exit")
		default:
		}
		params.Marker = resp.Marker  
		resp, err = c.RDS.DownloadDBLogFilePortion(params)
		if err != nil {
			return logFile, err
		}
		if _, err := io.WriteString(outfile, aws.StringValue(resp.LogFileData)); err != nil {
			return logFile, err
		}
	}
	return logFile, nil
}
func (c *CLI) GetLogFiles() ([]LogFile, error) {
	logFiles, err := c.getListRDSLogFiles()
	if err != nil {
		return nil, err
	}
	var matchingLogFiles []LogFile
	for _, lf := range logFiles {
		if strings.HasPrefix(lf.LogFileName, c.Options.LogFile) {
			matchingLogFiles = append(matchingLogFiles, lf)
		}
	}
	if len(matchingLogFiles) == 0 {
		errParts := []string{"No log file with the given prefix found. Available log files:"}
		for _, lf := range logFiles {
			errParts = append(errParts, fmt.Sprint("\t", lf.String()))
		}
		return nil, fmt.Errorf(strings.Join(errParts, "\n"))
	}
	return matchingLogFiles, nil
}
func (c *CLI) GetLatestLogFile() (LogFile, error) {
	logFiles, err := c.GetLogFiles()
	if err != nil {
		return LogFile{}, err
	}
	if len(logFiles) == 0 {
		return LogFile{}, errors.New("No log files found")
	}
	sort.SliceStable(logFiles, func(i, j int) bool { return logFiles[i].LastWritten < logFiles[j].LastWritten })
	return logFiles[len(logFiles)-1], nil
}