· 8 years ago · Jan 17, 2018, 05:22 PM
1package main
2
3// Parses classic ELB access logs and puts them inside a MySQL/MariaDB table
4// Easy way to get a DB:
5// docker run --name some-mariadb -e MYSQL_ROOT_PASSWORD=my-secret-pw -e MYSQL_DATABASE=accesslogs -p 3306:3306 -d mariadb:latest
6//
7// The bulk load your files:
8// for f in ~/Downloads/*.txt ; do go run aws_elb_log_analyzer.go -db-create-table -db-host "tcp(172.17.0.2)" -db-name accesslogs -db-user root -db-pwd my-secret-pw -db-table bla -file-path $f; done
9// Or:
10// TBL=bla ; find /tmp/${TBL} -type f -name '*.log' -o -name '*.txt' | while read f; do echo "Processing $f"; go run aws_elb_log_analyzer.go -db-create-table -db-host "tcp(172.17.0.2)" -db-name accesslogs -db-user root -db-pwd my-secret-pw -db-table ${TBL} -file-path $f; done
11//
12// And do reports:
13// - By day and IP
14// select year, month, day, sourceIP, count(*) as nbrcalls from bla group by year, month, day, sourceIP order by nbrcalls;
15// - By uri
16// select SUBSTRING_INDEX(uri, '?', 1), count(*) as nbrcalls from bla group by SUBSTRING_INDEX(uri, '?', 1) order by nbrcalls;
17// - By userAgent
18// select SUBSTRING_INDEX(userAgent, ' (', 1), count(*) as nbrcalls from bla group by SUBSTRING_INDEX(userAgent, ' (', 1) order by nbrcalls;
19// - A bit of filtering
20// select year, month, day, hour, SUBSTRING_INDEX(userAgent, ' (', 1) as agent, SUBSTRING_INDEX(uri, '?', 1) as uri, count(*) as nbrcalls from bla where userAgent not like 'Pingdom%' and userAgent != 'ZmEu' group by year, month, day, hour, SUBSTRING_INDEX(userAgent, ' (', 1), SUBSTRING_INDEX(uri, '?', 1) order by year, month, day, hour, nbrcalls;
21// Usage example in e a script:
22/*
23#!/bin/bash
24if [ $# -ne 1 ]; then
25 echo "argument required"
26 exit 1
27fi
28
29TBL=$1
30BUCKET=my-elb-logs
31aws s3 cp --recursive --exclude "*" --include "*2018/*" s3://${BUCKET}/${TBL}/AWSLogs/ /tmp/${TBL}
32find /tmp/${TBL} -type f -name '*.log' -o -name '*.txt' | while read f; do
33 echo "Processing $f"
34 go run aws_elb_log_analyzer.go -db-create-table -db-host "tcp(172.17.0.2)" -db-name accesslogs -db-user root -db-pwd my-secret-pw -db-table ${TBL} -file-path $f
35done
36mysql -h 172.17.0.2 -u root --password=my-secret-pw --database accesslogs -e "select CONCAT(year, '-', month, '-', day) as date, SUBSTRING_INDEX(userAgent, ' ', 1) as agent, SUBSTRING_INDEX(SUBSTRING_INDEX(REPLACE(uri,'//','/'), '?', 1), '/', 3) as shorturi, count(*) as nbrcalls from \`${TBL}\` where userAgent not like 'Pingdom%' and userAgent != 'ZmEu' group by year, month, day, SUBSTRING_INDEX(userAgent, ' ', 1), SUBSTRING_INDEX(SUBSTRING_INDEX(REPLACE(uri,'//','/'), '?', 1), '/', 3) order by year, month, day, nbrcalls" -B > /tmp/${TBL}_short.tsv
37
38echo "Requests per day" > /tmp/${TBL}_summary.tsv
39mysql -h 172.17.0.2 -u root --password=my-secret-pw --database accesslogs -e "select CONCAT(year, '-', month, '-', day) as date, count(*) as nbrcalls from \`${TBL}\` where userAgent not like 'Pingdom%' and userAgent != 'ZmEu' group by year, month, day order by year, month, day, nbrcalls" -B >> /tmp/${TBL}_summary.tsv
40echo "" >> /tmp/${TBL}_summary.tsv
41echo "Requests per method and scheme" >> /tmp/${TBL}_summary.tsv
42mysql -h 172.17.0.2 -u root --password=my-secret-pw --database accesslogs -e "select method, scheme, count(*) as nbrcalls from \`${TBL}\` where userAgent not like 'Pingdom%' and userAgent != 'ZmEu' group by method, scheme order by nbrcalls" -B >> /tmp/${TBL}_summary.tsv
43echo "" >> /tmp/${TBL}_summary.tsv
44echo "Top 10 source IP" >> /tmp/${TBL}_summary.tsv
45mysql -h 172.17.0.2 -u root --password=my-secret-pw --database accesslogs -e "select * from (select sourceIP, count(*) as nbrcalls from \`${TBL}\` where userAgent not like 'Pingdom%' and userAgent != 'ZmEu' group by sourceIP order by nbrcalls desc) t limit 10;" -B >> /tmp/${TBL}_summary.tsv
46echo "" >> /tmp/${TBL}_summary.tsv
47echo "Top 10 full user agent" >> /tmp/${TBL}_summary.tsv
48mysql -h 172.17.0.2 -u root --password=my-secret-pw --database accesslogs -e "select * from (select userAgent, count(*) as nbrcalls from \`${TBL}\` where userAgent not like 'Pingdom%' and userAgent != 'ZmEu' group by userAgent order by nbrcalls desc) t limit 10;" -B >> /tmp/${TBL}_summary.tsv
49echo "" >> /tmp/${TBL}_summary.tsv
50echo "Top 10 short user agent" >> /tmp/${TBL}_summary.tsv
51mysql -h 172.17.0.2 -u root --password=my-secret-pw --database accesslogs -e "select * from (select SUBSTRING_INDEX(SUBSTRING_INDEX(userAgent, ' ', 1),'(',1) as userAgent, count(*) as nbrcalls from \`${TBL}\` where userAgent not like 'Pingdom%' and userAgent != 'ZmEu' group by SUBSTRING_INDEX(SUBSTRING_INDEX(userAgent, ' ', 1),'(',1) order by nbrcalls desc) t limit 10;" -B >> /tmp/${TBL}_summary.tsv
52echo "" >> /tmp/${TBL}_summary.tsv
53echo "Top 10 root uri path" >> /tmp/${TBL}_summary.tsv
54mysql -h 172.17.0.2 -u root --password=my-secret-pw --database accesslogs -e "select * from (select SUBSTRING_INDEX(SUBSTRING_INDEX(REPLACE(uri,'//','/'), '?', 1), '/', 2) as root_uri, count(*) as nbrcalls from \`${TBL}\` where userAgent not like 'Pingdom%' and userAgent != 'ZmEu' group by SUBSTRING_INDEX(SUBSTRING_INDEX(REPLACE(uri,'//','/'), '?', 1), '/', 2) order by nbrcalls desc) t limit 10;" -B >> /tmp/${TBL}_summary.tsv
55echo "" >> /tmp/${TBL}_summary.tsv
56echo "Top 10 short uri path" >> /tmp/${TBL}_summary.tsv
57mysql -h 172.17.0.2 -u root --password=my-secret-pw --database accesslogs -e "select * from (select SUBSTRING_INDEX(SUBSTRING_INDEX(REPLACE(uri,'//','/'), '?', 1), '/', 3) as short_uri, count(*) as nbrcalls from \`${TBL}\` where userAgent not like 'Pingdom%' and userAgent != 'ZmEu' group by SUBSTRING_INDEX(SUBSTRING_INDEX(REPLACE(uri,'//','/'), '?', 1), '/', 3) order by nbrcalls desc) t limit 10;" -B >> /tmp/${TBL}_summary.tsv
58echo "" >> /tmp/${TBL}_summary.tsv
59echo "Top 10 raw uri path" >> /tmp/${TBL}_summary.tsv
60mysql -h 172.17.0.2 -u root --password=my-secret-pw --database accesslogs -e "select * from (select SUBSTRING_INDEX(uri,'?', 1) as uri, count(*) as nbrcalls from \`${TBL}\` where userAgent not like 'Pingdom%' and userAgent != 'ZmEu' group by SUBSTRING_INDEX(uri, '?', 1) order by nbrcalls desc) t limit 10;" -B >> /tmp/${TBL}_summary.tsv
61rm -rf /tmp/${TBL}/
62
63*/
64
65import (
66 "bufio"
67 "flag"
68 "fmt"
69 "net/url"
70 "os"
71 "regexp"
72 "sync"
73 "time"
74
75 "database/sql"
76 _ "github.com/go-sql-driver/mysql"
77 "github.com/gobike/envflag"
78)
79
80var wg sync.WaitGroup
81
82type accessLogEntry struct {
83 year, month, day, hour int
84 sourceIP, method, domain, scheme, uri, userAgent string
85}
86
87// processLine takes a line and the compiled regex and returns a accessLogEntry
88func processLine(re *regexp.Regexp, line string) *accessLogEntry {
89 entry := accessLogEntry{}
90
91 result := re.FindStringSubmatch(line)
92 // fmt.Printf("%s --->>> %v\n", line, result)
93
94 // do not process incorrect lines
95 if len(result) < 18 {
96 return nil
97 }
98 layout := "2006-01-02T15:04:05.000000Z"
99 mDate, err := time.Parse(layout, result[1])
100 if err != nil {
101 fmt.Println(err)
102 }
103 entry.year = mDate.Year()
104 entry.month = int(mDate.Month())
105 entry.day = mDate.Day()
106 entry.hour = mDate.Hour()
107
108 entry.sourceIP = result[3]
109 entry.method = result[14]
110
111 u, err := url.Parse(result[15])
112 if err != nil {
113 fmt.Println(err)
114 } else {
115 entry.domain = u.Hostname()
116 entry.scheme = u.Scheme
117 entry.uri = u.RequestURI()
118 }
119
120 entry.userAgent = result[17]
121 // fmt.Println(entry)
122 return &entry
123}
124
125// processFile reads a file and process each of the lines and sends them to the
126// given open channel
127func processFile(path string, dataPipe chan *accessLogEntry) {
128 inFile, _ := os.Open(path)
129 defer inFile.Close()
130 scanner := bufio.NewScanner(inFile)
131 scanner.Split(bufio.ScanLines)
132
133 pattern := regexp.MustCompile(`^([^ ]*) ([^ ]*) ([^ ]*):([0-9]*) ([^ ]*)[:\-]([0-9]*) ([-.0-9]*) ([-.0-9]*) ([-.0-9]*) (|[-0-9]*) (-|[-0-9]*) ([-0-9]*) ([-0-9]*) "([^ ]*) ([^ ]*) (- |[^ ]*)" "([^"]*)" ([A-Z0-9-]+) ([A-Za-z0-9.-]*)$`)
134 for scanner.Scan() {
135 dataPipe <- processLine(pattern, scanner.Text())
136 }
137 close(dataPipe)
138}
139
140// Takes the data out of the given channel and pushes it to the given mysql
141// table
142func channelToDB(user, pwd, host, database, tableName string, createTbl bool, dataPipe chan *accessLogEntry) {
143 db, err := sql.Open("mysql", fmt.Sprintf("%s:%s@%s/%s?charset=utf8", user, pwd, host, database))
144 if err != nil {
145 panic(err)
146 }
147 defer db.Close()
148
149 if createTbl {
150 crStmt, err := db.Prepare(fmt.Sprintf("CREATE TABLE IF NOT EXISTS `%s` (`year` INT(4), `month` INT(2), `day` INT(2), `hour` INT(2), `sourceIP` VARCHAR(128), `method` VARCHAR(8), `domain` VARCHAR(256), `scheme` VARCHAR(8), `uri` VARCHAR(512), `userAgent` VARCHAR(512))", tableName))
151 if err != nil {
152 fmt.Println(err)
153 }
154
155 _, err = crStmt.Exec()
156 if err != nil {
157 fmt.Println(err)
158 }
159 crStmt.Close()
160 }
161
162 var (
163 tx *sql.Tx
164 stmt *sql.Stmt
165 )
166 flagIdx := 0
167 for elem := range dataPipe {
168
169 if elem == nil {
170 continue
171 }
172 if flagIdx == 0 {
173 tx, err = db.Begin()
174 if err != nil {
175 fmt.Println(err)
176 }
177 stmt, err = tx.Prepare(fmt.Sprintf("insert into `%s` (`year`, `month`, `day`, `hour`, `sourceIP`, `method`, `domain`, `scheme`, `uri`, `userAgent`) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", tableName))
178 if err != nil {
179 fmt.Println(err)
180 }
181 defer stmt.Close()
182 }
183
184 uriLen := len(elem.uri)
185 if uriLen > 511 {
186 uriLen = 511
187 }
188 agentLen := len(elem.userAgent)
189 if agentLen > 511 {
190 agentLen = 511
191 }
192 _, err = stmt.Exec(elem.year, elem.month, elem.day, elem.hour, elem.sourceIP, elem.method, elem.domain, elem.scheme, elem.uri[:uriLen], elem.userAgent[:agentLen])
193 if err != nil {
194 fmt.Println(err)
195 }
196
197 flagIdx++
198 if flagIdx > 10000 {
199 err = tx.Commit()
200 if err != nil {
201 fmt.Println(err)
202 }
203 flagIdx = 0
204 }
205 }
206 err = tx.Commit()
207 if err != nil {
208 fmt.Println(err)
209 }
210
211 wg.Done()
212}
213
214func main() {
215 var (
216 fPath, dbName, dbHost, dbUser, dbPassword, dbTable string
217 dbCreateTable bool
218 )
219 flag.StringVar(&fPath, "file-path", "text", "Path to the log file. Environment variable: FILE_PATH")
220 flag.StringVar(&dbName, "db-name", "accesslogs", "Name of the DB to connect to. Environment variable: DB_NAME")
221 flag.StringVar(&dbHost, "db-host", "", "Name of the DB server to connect to. Environment variable: DB_HOST")
222 flag.StringVar(&dbUser, "db-user", "", "User name to use to connect to the DB. Environment variable: DB_USER")
223 flag.StringVar(&dbPassword, "db-pwd", "", "Password to use to connect to the DB. Environment variable: DB_PWD")
224 flag.StringVar(&dbTable, "db-table", "", "Name of the table to import the data in. Environment variable: DB_TABLE")
225 flag.BoolVar(&dbCreateTable, "db-create-table", false, "Whether to create the table if it does not exists. Environment variable: DB_CREATE_TABLE")
226 envflag.Parse()
227
228 dp := make(chan *accessLogEntry)
229 wg.Add(1)
230 go channelToDB(dbUser, dbPassword, dbHost, dbName, dbTable, dbCreateTable, dp)
231
232 processFile(fPath, dp)
233 wg.Wait()
234}