<commit_before>package main

import (
	"bufio"
	"bytes"
	"encoding/base64"
	"errors"
	"log"
	"net/http"
	"strings"
	"time"

	"github.com/bmizerany/lpx"
	"github.com/kr/logfmt"
	metrics "github.com/rcrowley/go-metrics"
)

var (
	TokenPrefix = []byte("t.")
	Heroku      = []byte("heroku")

	// go-metrics Instruments
	wrongMethodErrorCounter    = metrics.NewRegisteredCounter("lumbermill.errors.drain.wrong.method", metrics.DefaultRegistry)
	authFailureCounter         = metrics.NewRegisteredCounter("lumbermill.errors.auth.failure", metrics.DefaultRegistry)
	badRequestCounter          = metrics.NewRegisteredCounter("lumbermill.errors.badrequest", metrics.DefaultRegistry)
	internalServerErrorCounter = metrics.NewRegisteredCounter("lumbermill.errors.internalserver", metrics.DefaultRegistry)
	tokenMissingCounter        = metrics.NewRegisteredCounter("lumbermill.errors.token.missing", metrics.DefaultRegistry)
	timeParsingErrorCounter    = metrics.NewRegisteredCounter("lumbermill.errors.time.parse", metrics.DefaultRegistry)
	logfmtParsingErrorCounter  = metrics.NewRegisteredCounter("lumbermill.errors.logfmt.parse", metrics.DefaultRegistry)
	droppedErrorCounter        = metrics.NewRegisteredCounter("lumbermill.errors.dropped", metrics.DefaultRegistry)
	batchCounter               = metrics.NewRegisteredCounter("lumbermill.batch", metrics.DefaultRegistry)
	linesCounter               = metrics.NewRegisteredCounter("lumbermill.lines", metrics.DefaultRegistry)
	routerErrorLinesCounter    = metrics.NewRegisteredCounter("lumbermill.lines.router.error", metrics.DefaultRegistry)
	routerLinesCounter         = metrics.NewRegisteredCounter("lumbermill.lines.router", metrics.DefaultRegistry)
	dynoErrorLinesCounter      = metrics.NewRegisteredCounter("lumbermill.lines.dyno.error", metrics.DefaultRegistry)
	dynoMemLinesCounter        = metrics.NewRegisteredCounter("lumbermill.lines.dyno.mem", metrics.DefaultRegistry)
	dynoLoadLinesCounter       = metrics.NewRegisteredCounter("lumbermill.lines.dyno.load", metrics.DefaultRegistry)
	unknownHerokuLinesCounter  = metrics.NewRegisteredCounter("lumbermill.lines.unknown.heroku", metrics.DefaultRegistry)
	unknownUserLinesCounter    = metrics.NewRegisteredCounter("lumbermill.lines.unknown.user", metrics.DefaultRegistry)
	parseTimer                 = metrics.NewRegisteredTimer("lumbermill.batches.parse.time", metrics.DefaultRegistry)
	batchSizeHistogram         = metrics.NewRegisteredHistogram("lumbermill.batches.sizes", metrics.DefaultRegistry, metrics.NewUniformSample(100))
)

func checkAuth(r *http.Request) error {
	header := r.Header.Get("Authorization")
	if header == "" {
		return errors.New("Authorization required")
	}
	headerParts := strings.SplitN(header, " ", 2)
	if len(headerParts) != 2 {
		return errors.New("Authorization header is malformed")
	}

	method := headerParts[0]
	if method != "Basic" {
		return errors.New("Only Basic Authorization is accepted")
	}

	encodedUserPass := headerParts[1]
	decodedUserPass, err := base64.StdEncoding.DecodeString(encodedUserPass)
	if err != nil {
		return errors.New("Authorization header is malformed")
	}

	userPassParts := bytes.SplitN(decodedUserPass, []byte{':'}, 2)
	if len(userPassParts) != 2 {
		return errors.New("Authorization header is malformed")
	}

	user := userPassParts[0]
	pass := userPassParts[1]

	if string(user) != User {
		return errors.New("Unknown user")
	}
	if string(pass) != Password {
		return errors.New("Incorrect token")
	}

	return nil
}

// Dyno's are generally reported as "<type>.<#>"
// Extract the <type> and return it
func dynoType(what string) string {
	s := strings.Split(what, ".")
	return s[0]
}

func handleLogFmtParsingError(msg []byte, err error) {
	logfmtParsingErrorCounter.Inc(1)
	log.Printf("logfmt unmarshal error(%q): %q\n", string(msg), err)
}

// "Parse tree" from hell
func serveDrain(w http.ResponseWriter, r *http.Request) {

	w.Header().Set("Content-Length", "0")

	if r.Method != "POST" {
		w.WriteHeader(http.StatusMethodNotAllowed)
		wrongMethodErrorCounter.Inc(1)
		return
	}

	id := r.Header.Get("Logplex-Drain-Token")

	if id == "" {
		if err := checkAuth(r); err != nil {
			w.WriteHeader(http.StatusForbidden)
			authFailureCounter.Inc(1)
			return
		}
	}

	batchCounter.Inc(1)

	parseStart := time.Now()
	lp := lpx.NewReader(bufio.NewReader(r.Body))

	linesCounterInc := 0

	for lp.Next() {
		linesCounterInc += 1
		header := lp.Header()

		// If the syslog Name Header field contains what looks like a log token,
		// let's assume it's an override of the id and we're getting the data from the magic
		// channel
		if bytes.HasPrefix(header.Name, TokenPrefix) {
			id = string(header.Name)
		}

		// If we still don't have an id, throw an error and try the next line
		if id == "" {
			tokenMissingCounter.Inc(1)
			continue
		}

		destination := hashRing.Get(id)

		msg := lp.Bytes()
		switch {
		case bytes.Equal(header.Name, Heroku), bytes.HasPrefix(header.Name, TokenPrefix):
			timeStr := string(lp.Header().Time)
			t, e := time.Parse("2006-01-02T15:04:05.000000+00:00", timeStr)
			if e != nil {
				t, e = time.Parse("2006-01-02T15:04:05+00:00", timeStr)
				if e != nil {
					timeParsingErrorCounter.Inc(1)
					log.Printf("Error Parsing Time(%s): %q\n", string(lp.Header().Time), e)
					continue
				}
			}

			timestamp := t.UnixNano() / int64(time.Microsecond)

			pid := string(header.Procid)
			switch pid {
			case "router":

				switch {
				// router logs with a H error code in them
				case bytes.Contains(msg, keyCodeH):
					routerErrorLinesCounter.Inc(1)
					re := routerError{}
					err := logfmt.Unmarshal(msg, &re)
					if err != nil {
						handleLogFmtParsingError(msg, err)
						continue
					}
					destination.PostPoint(Point{id, EventsRouter, []interface{}{timestamp, re.Code}})

				//If the app is blank (not pushed) we don't care
				case bytes.Contains(msg, keyCodeBlank), bytes.Contains(msg, keyDescBlank):
					// do nothing

				// likely a standard router log
				default:
					routerLinesCounter.Inc(1)
					rm := routerMsg{}
					err := logfmt.Unmarshal(msg, &rm)
					if err != nil {
						handleLogFmtParsingError(msg, err)<commit_msg>Report a metric instead of doing nothing<commit_after>package main

import (
	"bufio"
	"bytes"
	"encoding/base64"
	"errors"
	"log"
	"net/http"
	"strings"
	"time"

	"github.com/bmizerany/lpx"
	"github.com/kr/logfmt"
	metrics "github.com/rcrowley/go-metrics"
)

var (
	TokenPrefix = []byte("t.")
	Heroku      = []byte("heroku")

	// go-metrics Instruments
	wrongMethodErrorCounter    = metrics.NewRegisteredCounter("lumbermill.errors.drain.wrong.method", metrics.DefaultRegistry)
	authFailureCounter         = metrics.NewRegisteredCounter("lumbermill.errors.auth.failure", metrics.DefaultRegistry)
	badRequestCounter          = metrics.NewRegisteredCounter("lumbermill.errors.badrequest", metrics.DefaultRegistry)
	internalServerErrorCounter = metrics.NewRegisteredCounter("lumbermill.errors.internalserver", metrics.DefaultRegistry)
	tokenMissingCounter        = metrics.NewRegisteredCounter("lumbermill.errors.token.missing", metrics.DefaultRegistry)
	timeParsingErrorCounter    = metrics.NewRegisteredCounter("lumbermill.errors.time.parse", metrics.DefaultRegistry)
	logfmtParsingErrorCounter  = metrics.NewRegisteredCounter("lumbermill.errors.logfmt.parse", metrics.DefaultRegistry)
	droppedErrorCounter        = metrics.NewRegisteredCounter("lumbermill.errors.dropped", metrics.DefaultRegistry)
	batchCounter               = metrics.NewRegisteredCounter("lumbermill.batch", metrics.DefaultRegistry)
	linesCounter               = metrics.NewRegisteredCounter("lumbermill.lines", metrics.DefaultRegistry)
	routerErrorLinesCounter    = metrics.NewRegisteredCounter("lumbermill.lines.router.error", metrics.DefaultRegistry)
	routerLinesCounter         = metrics.NewRegisteredCounter("lumbermill.lines.router", metrics.DefaultRegistry)
	routerBlankLinesCounter    = metrics.NewRegisteredCounter("lumbermill.lines.router.blank", metrics.DefaultRegistry)
	dynoErrorLinesCounter      = metrics.NewRegisteredCounter("lumbermill.lines.dyno.error", metrics.DefaultRegistry)
	dynoMemLinesCounter        = metrics.NewRegisteredCounter("lumbermill.lines.dyno.mem", metrics.DefaultRegistry)
	dynoLoadLinesCounter       = metrics.NewRegisteredCounter("lumbermill.lines.dyno.load", metrics.DefaultRegistry)
	unknownHerokuLinesCounter  = metrics.NewRegisteredCounter("lumbermill.lines.unknown.heroku", metrics.DefaultRegistry)
	unknownUserLinesCounter    = metrics.NewRegisteredCounter("lumbermill.lines.unknown.user", metrics.DefaultRegistry)
	parseTimer                 = metrics.NewRegisteredTimer("lumbermill.batches.parse.time", metrics.DefaultRegistry)
	batchSizeHistogram         = metrics.NewRegisteredHistogram("lumbermill.batches.sizes", metrics.DefaultRegistry, metrics.NewUniformSample(100))
)

func checkAuth(r *http.Request) error {
	header := r.Header.Get("Authorization")
	if header == "" {
		return errors.New("Authorization required")
	}
	headerParts := strings.SplitN(header, " ", 2)
	if len(headerParts) != 2 {
		return errors.New("Authorization header is malformed")
	}

	method := headerParts[0]
	if method != "Basic" {
		return errors.New("Only Basic Authorization is accepted")
	}

	encodedUserPass := headerParts[1]
	decodedUserPass, err := base64.StdEncoding.DecodeString(encodedUserPass)
	if err != nil {
		return errors.New("Authorization header is malformed")
	}

	userPassParts := bytes.SplitN(decodedUserPass, []byte{':'}, 2)
	if len(userPassParts) != 2 {
		return errors.New("Authorization header is malformed")
	}

	user := userPassParts[0]
	pass := <PASSWORD>Parts[1]

	if string(user) != User {
		return errors.New("Unknown user")
	}
	if string(pass) != Password {
		return errors.New("Incorrect token")
	}

	return nil
}

// Dyno's are generally reported as "<type>.<#>"
// Extract the <type> and return it
func dynoType(what string) string {
	s := strings.Split(what, ".")
	return s[0]
}

func handleLogFmtParsingError(msg []byte, err error) {
	logfmtParsingErrorCounter.Inc(1)
	log.Printf("logfmt unmarshal error(%q): %q\n", string(msg), err)
}

// "Parse tree" from hell
func serveDrain(w http.ResponseWriter, r *http.Request) {

	w.Header().Set("Content-Length", "0")

	if r.Method != "POST" {
		w.WriteHeader(http.StatusMethodNotAllowed)
		wrongMethodErrorCounter.Inc(1)
		return
	}

	id := r.Header.Get("Logplex-Drain-Token")

	if id == "" {
		if err := checkAuth(r); err != nil {
			w.WriteHeader(http.StatusForbidden)
			authFailureCounter.Inc(1)
			return
		}
	}

	batchCounter.Inc(1)

	parseStart := time.Now()
	lp := lpx.NewReader(bufio.NewReader(r.Body))

	linesCounterInc := 0

	for lp.Next() {
		linesCounterInc += 1
		header := lp.Header()

		// If the syslog Name Header field contains what looks like a log token,
		// let's assume it's an override of the id and we're getting the data from the magic
		// channel
		if bytes.HasPrefix(header.Name, TokenPrefix) {
			id = string(header.Name)
		}

		// If we still don't have an id, throw an error and try the next line
		if id == "" {
			tokenMissingCounter.Inc(1)
			continue
		}

		destination := hashRing.Get(id)

		msg := lp.Bytes()
		switch {
		case bytes.Equal(header.Name, Heroku), bytes.HasPrefix(header.Name, TokenPrefix):
			timeStr := string(lp.Header().Time)
			t, e := time.Parse("2006-01-02T15:04:05.000000+00:00", timeStr)
			if e != nil {
				t, e = time.Parse("2006-01-02T15:04:05+00:00", timeStr)
				if e != nil {
					timeParsingErrorCounter.Inc(1)
					log.Printf("Error Parsing Time(%s): %q\n", string(lp.Header().Time), e)
					continue
				}
			}

			timestamp := t.UnixNano() / int64(time.Microsecond)

			pid := string(header.Procid)
			switch pid {
			case "router":

				switch {
				// router logs with a H error code in them
				case bytes.Contains(msg, keyCodeH):
					routerErrorLinesCounter.Inc(1)
					re := routerError{}
					err := logfmt.Unmarshal(msg, &re)
					if err != nil {
						handleLogFmtParsingError(msg, err)
						continue
					}
					destination.PostPoint(Point{id, EventsRouter, []interface{}{timestamp, re.Code}})

				// If the app is blank (not pushed) we don't care
				// do nothing atm, increment a counter
				case bytes.Contains(msg, keyCodeBlank), bytes.Contains(msg, keyDescBlank):
					routerBlankLinesCounter.Inc(1)

				// likely a standard router log
				default:
					routerLinesCounter.Inc(1)
					rm := routerMsg{}
					err := logfmt.Unmarshal(msg, &rm)
					if err != nil {
						handleLogFmtParsingError(msg, err)
{'fraction_non_alphanumeric': 0.12582781456953643, 'fraction_numerical': 0.011409877922285167, 'mean_word_length': 4.26417471650567, 'pattern_counts': {'":': 2, '<': 10, '<?xml version=': 0, '>': 10, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, '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': '7329264', 'n_tokens_mistral': 4389, 'n_tokens_neox': 3914, 'n_words': 793}