<commit_before>// Package mongodb is a parser for mongodb logs
package mongodb

import (
	"math/rand"
	"time"

	"github.com/Sirupsen/logrus"
	"github.com/tmc/mongologtools/parser"

	"github.com/honeycombio/honeytail/event"
)

type Options struct {
}

type Parser struct {
	conf       Options
	lineParser LineParser
	nower      Nower
}

type LineParser interface {
	ParseLogLine(line string) (map[string]interface{}, error)
}

type MongoLineParser struct {
}

func (m *MongoLineParser) ParseLogLine(line string) (map[string]interface{}, error) {
	return parser.ParseLogLine(line)
}

func (p *Parser) Init(_ interface{}) error {
	p.nower = &RealNower{}
	p.lineParser = &MongoLineParser{}
	return nil
}

func (p *Parser) ProcessLines(lines <-chan string, send chan<- event.Event) {
	for line := range lines {
		values, err := p.lineParser.ParseLogLine(line)
		// we get a bunch of errors from the parser on mongo logs, skip em
		if err == nil {
			logrus.WithFields(logrus.Fields{
				"line":   line,
				"values": values,
			}).Debug("Successfully parsed line")
			// for each entry, make a json blob with key/value pairs for each value map
			e := event.Event{
				Timestamp: randomTime(p.nower),
				Data:      values,
			}
			send <- e
		} else {
			logrus.WithFields(logrus.Fields{
				"line": line,
			}).Debug("logline didn't parse, skipping.")
		}
	}
	logrus.Debug("lines channel is closed, ending mongo processor")
}

type Nower interface {
	Now() time.Time
}

type RealNower struct{}

func (r *RealNower) Now() time.Time {
	return time.Now().UTC()
}

func randomTime(n Nower) time.Time {
	return n.Now().Add(-1 * time.Duration(rand.Int63n(604800)) * time.Second)
}<commit_msg>use the new log parser for mongodb<commit_after>// Package mongodb is a parser for mongodb logs
package mongodb

import (
	"errors"
	"time"

	"github.com/Sirupsen/logrus"
	"github.com/honeycombio/mongodbtools/logparser"

	"github.com/honeycombio/honeytail/event"
)

const (
	ctimeTimeFormat        = "Mon Jan _2 15:04:05.000"
	ctimeNoMSTimeFormat    = "Mon Jan _2 15:04:05"
	iso8601UTCTimeFormat   = "2006-01-02T15:04:05Z"
	iso8601LocalTimeFormat = "2006-01-02T15:04:05.999999999-0700"
)

var timestampFormats = []string{iso8601LocalTimeFormat, iso8601UTCTimeFormat, ctimeNoMSTimeFormat, ctimeTimeFormat}

type Options struct {
	LogPartials bool `long:"log_partials" description:"Send what was successfully parsed from a line (only if the error occured in the log line's message)."`
}

type Parser struct {
	conf       Options
	lineParser LineParser
	nower      Nower
}

type LineParser interface {
	ParseLogLine(line string) (map[string]interface{}, error)
}

type MongoLineParser struct {
}

func (m *MongoLineParser) ParseLogLine(line string) (map[string]interface{}, error) {
	return logparser.ParseLogLine(line)
}

func (p *Parser) Init(options interface{}) error {
	p.conf = *options.(*Options)
	p.nower = &RealNower{}
	p.lineParser = &MongoLineParser{}
	return nil
}

func (p *Parser) parseTimestamp(values map[string]interface{}) (time.Time, error) {
	now := p.nower.Now()
	timestamp_value, ok := values["timestamp"].(string)
	if ok {
		var err error
		for _, f := range timestampFormats {
			timestamp, err := time.Parse(f, timestamp_value)
			if err == nil {
				if f == ctimeTimeFormat || f == ctimeNoMSTimeFormat {
					// these formats lacks the year, so we check
					// if adding Now().Year causes the date to be
					// after today.  if it's after today, we
					// decrement year by 1.  if it's not after, we
					// use it.
					ts := timestamp.AddDate(now.Year(), 0, 0)
					if now.After(ts) {
						return ts, nil
					}

					return timestamp.AddDate(now.Year()-1, 0, 0), nil
				}
				return timestamp, nil
			}
		}
		return time.Time{}, err
	}

	return time.Time{}, errors.New("timestamp missing from logline")
}

func (p *Parser) ProcessLines(lines <-chan string, send chan<- event.Event) {
	for line := range lines {
		values, err := p.lineParser.ParseLogLine(line)
		// we get a bunch of errors from the parser on mongo logs, skip em
		if err == nil || (p.conf.LogPartials && logparser.IsPartialLogLine(err)) {
			timestamp, err := p.parseTimestamp(values)
			if err != nil {
				logrus.WithFields(logrus.Fields{
					"line": line,
				}).WithError(err).Debug("couldn't parse logline timestamp, skipping.")
				continue
			}
			logrus.WithFields(logrus.Fields{
				"line":   line,
				"values": values,
			}).Debug("Successfully parsed line")

			// we'll be putting the timestamp in the Event
			// itself, no need to also have it in the Data
			delete(values, "timestamp")

			send <- event.Event{
				Timestamp: timestamp,
				Data:      values,
			}
		} else {
			logrus.WithFields(logrus.Fields{
				"line": line,
			}).WithError(err).Debug("logline didn't parse, skipping.")
		}
	}
	logrus.Debug("lines channel is closed, ending mongo processor")
}

type Nower interface {
	Now() time.Time
}

type RealNower struct{}

func (r *RealNower) Now() time.Time {
	return time.Now().UTC()
}
{'fraction_non_alphanumeric': 0.12707624574744847, 'fraction_numerical': 0.017810686411847108, 'mean_word_length': 3.623496762257169, 'pattern_counts': {'":': 7, '<': 9, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 1, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6407483', 'n_tokens_mistral': 1923, 'n_tokens_neox': 1680, 'n_words': 478}