munchclient/apache.go

package main

import (
	"bufio"
	"log"
	"net"
	"os"
	"strings"
	"time"

	"git.scraperwall.com/scw/data/v2"
	"github.com/Songmu/axslogparser"
	"github.com/hpcloud/tail"
)

func apacheLogReplay(logfile string) {
	file, err := os.Open(logfile)
	if err != nil {
		log.Fatalf("%s: %s", logfile, err)
	}
	defer file.Close()

	scanner := bufio.NewScanner(file)

	var p axslogparser.Parser
	parserSet := false

	var tOffset time.Duration

	for scanner.Scan() {
		l := scanner.Text()

		if err := scanner.Err(); err != nil {
			log.Fatal(err)
		}

		if !parserSet {
			p, _, err = axslogparser.GuessParser(l)
			if err != nil {
				log.Println(err)
				continue
			}
			parserSet = true
		}

		logEntry, err := p.Parse(l)
		if err != nil {
			log.Println(err)
			continue
		}

		if tOffset == 0 {
			tOffset = time.Now().Sub(logEntry.Time)
		}

		ts := logEntry.Time.Add(tOffset)
		if ts.After(time.Now()) {
			time.Sleep(ts.Sub(time.Now()))
		}

		remote := logEntry.Host

		if logEntry.VirtualHost != "" && config.UseVhostAsSource {
			commaIdx := strings.Index(logEntry.VirtualHost, ",")
			if commaIdx > 0 {
				logEntry.VirtualHost = logEntry.VirtualHost[0:commaIdx]
			}
			vhIP := net.ParseIP(logEntry.VirtualHost)
			if vhIP != nil {
				remote = logEntry.VirtualHost
			}
		}
		if *useXForwardedAsSource && logEntry.ForwardedFor != "" {
			remote = logEntry.ForwardedFor
		}

		// only use the first host in case there are multiple hosts in the log
		if cidx := strings.Index(remote, ","); cidx >= 0 {
			remote = remote[0:cidx]
		}

		// extract the virtual host
		var virtualHost string
		vhost := logEntry.VirtualHost
		if vhost != "" {
			vhostAndPort := strings.Split(logEntry.VirtualHost, ":")
			virtualHost = vhostAndPort[0]

		} else {
			if config.HostName != "" {
				vhost = config.HostName
			} else {
				vhost = "[not available]"
			}
		}

		request := data.Request{
			IpSrc:         remote,
			IpDst:         "127.0.0.1",
			PortSrc:       0,
			PortDst:       0,
			TcpSeq:        0,
			CreatedAt:     (logEntry.Time.Add(tOffset)).UnixNano(),
			Url:           logEntry.RequestURI,
			Method:        logEntry.Method,
			Host:          virtualHost,
			Protocol:      logEntry.Protocol,
			Origin:        remote,
			Source:        remote,
			Referer:       logEntry.Referer,
			XForwardedFor: logEntry.ForwardedFor,
			UserAgent:     logEntry.UserAgent,
		}

		if config.Trace {
			log.Printf("[%s] %s\n", request.Source, request.Url)
		}

		count++
		publishRequest(config.NatsQueue, &request)

	}
}

func apacheLogCapture(logfile string) {
	if _, err := os.Stat(logfile); err != nil {
		log.Fatalf("%s: %s", logfile, err)
	}

	t, err := tail.TailFile(logfile, tail.Config{
		Follow:   true,                                 // follow the file
		ReOpen:   true,                                 // reopen log file when it gets closed/rotated
		Logger:   tail.DiscardingLogger,                // don't log anything
		Poll:     config.TailPoll,                      // use file polling to detect a file rollover
		Location: &tail.SeekInfo{Offset: 0, Whence: 2}, // start at the end of the file
	})
	if err != nil {
		log.Fatalf("%s: %s", logfile, err)
	}

	var p axslogparser.Parser
	parserSet := false

	for line := range t.Lines {
		l := line.Text

		if !parserSet {
			p, _, err = axslogparser.GuessParser(l)
			if err != nil {
				log.Println(err)
				continue
			}
			parserSet = true
		}

		logEntry, err := p.Parse(l)
		if err != nil {
			log.Println(err)
			continue
		}

		remote := logEntry.Host
		if config.UseXForwardedAsSource && logEntry.ForwardedFor != "" {
			remote = logEntry.ForwardedFor
		}

		// only use the first host in case there are multiple hosts in the log
		if cidx := strings.Index(remote, ","); cidx >= 0 {
			remote = remote[0:cidx]
		}

		// extract the virtual host
		var virtualHost string
		vhost := logEntry.VirtualHost
		if vhost != "" {
			vhostAndPort := strings.Split(vhost, ":")
			virtualHost = vhostAndPort[0]

			if config.UseVhostAsSource {
				vhosts := strings.Split(vhost, ",")
				virtualHost = strings.TrimSpace(vhosts[0])
				vhIP := net.ParseIP(virtualHost)
				if vhIP != nil {
					remote = virtualHost
				}
			}
		} else {
			if config.HostName != "" {
				vhost = config.HostName
			} else {
				vhost = "[not available]"
			}
		}

		request := data.Request{
			IpSrc:         remote,
			IpDst:         "127.0.0.1",
			PortSrc:       0,
			PortDst:       0,
			TcpSeq:        0,
			CreatedAt:     logEntry.Time.UnixNano(),
			Url:           logEntry.RequestURI,
			Method:        logEntry.Method,
			Host:          virtualHost,
			Protocol:      logEntry.Protocol,
			Origin:        remote,
			Source:        remote,
			Referer:       logEntry.Referer,
			XForwardedFor: logEntry.ForwardedFor,
			UserAgent:     logEntry.UserAgent,
		}

		if config.Trace {
			log.Printf("[%s] %s\n", request.Source, request.Url)
		}

		count++
		publishRequest(config.NatsQueue, &request)
	}
}