apache.go 4.7 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218
  1. package main
  2. import (
  3. "bufio"
  4. "log"
  5. "net"
  6. "os"
  7. "strings"
  8. "time"
  9. "git.scraperwall.com/scw/data"
  10. "github.com/Songmu/axslogparser"
  11. "github.com/hpcloud/tail"
  12. )
  13. func apacheLogReplay(logfile string) {
  14. file, err := os.Open(logfile)
  15. if err != nil {
  16. log.Fatalf("%s: %s", logfile, err)
  17. }
  18. defer file.Close()
  19. scanner := bufio.NewScanner(file)
  20. var p axslogparser.Parser
  21. parserSet := false
  22. var tOffset time.Duration
  23. for scanner.Scan() {
  24. l := scanner.Text()
  25. if err := scanner.Err(); err != nil {
  26. log.Fatal(err)
  27. }
  28. if !parserSet {
  29. p, _, err = axslogparser.GuessParser(l)
  30. if err != nil {
  31. log.Println(err)
  32. continue
  33. }
  34. parserSet = true
  35. }
  36. logEntry, err := p.Parse(l)
  37. if err != nil {
  38. log.Println(err)
  39. continue
  40. }
  41. if tOffset == 0 {
  42. tOffset = time.Now().Sub(logEntry.Time)
  43. }
  44. ts := logEntry.Time.Add(tOffset)
  45. if ts.After(time.Now()) {
  46. time.Sleep(ts.Sub(time.Now()))
  47. }
  48. remote := logEntry.Host
  49. if logEntry.VirtualHost != "" && config.UseVhostAsSource {
  50. commaIdx := strings.Index(logEntry.VirtualHost, ",")
  51. if commaIdx > 0 {
  52. logEntry.VirtualHost = logEntry.VirtualHost[0:commaIdx]
  53. }
  54. vhIP := net.ParseIP(logEntry.VirtualHost)
  55. if vhIP != nil {
  56. remote = logEntry.VirtualHost
  57. }
  58. }
  59. if *useXForwardedAsSource && logEntry.ForwardedFor != "" {
  60. remote = logEntry.ForwardedFor
  61. }
  62. // only use the first host in case there are multiple hosts in the log
  63. if cidx := strings.Index(remote, ","); cidx >= 0 {
  64. remote = remote[0:cidx]
  65. }
  66. // extract the virtual host
  67. var virtualHost string
  68. vhost := logEntry.VirtualHost
  69. if vhost != "" {
  70. vhostAndPort := strings.Split(logEntry.VirtualHost, ":")
  71. virtualHost = vhostAndPort[0]
  72. } else {
  73. if config.HostName != "" {
  74. vhost = config.HostName
  75. } else {
  76. vhost = "[not available]"
  77. }
  78. }
  79. request := data.Request{
  80. IpSrc: remote,
  81. IpDst: "127.0.0.1",
  82. PortSrc: 0,
  83. PortDst: 0,
  84. TcpSeq: 0,
  85. CreatedAt: (logEntry.Time.Add(tOffset)).UnixNano(),
  86. Url: logEntry.RequestURI,
  87. Method: logEntry.Method,
  88. Host: virtualHost,
  89. Protocol: logEntry.Protocol,
  90. Origin: remote,
  91. Source: remote,
  92. Referer: logEntry.Referer,
  93. XForwardedFor: logEntry.ForwardedFor,
  94. UserAgent: logEntry.UserAgent,
  95. }
  96. if config.Trace {
  97. log.Printf("[%s] %s\n", request.Source, request.Url)
  98. }
  99. count++
  100. publishRequest(config.NatsQueue, &request)
  101. }
  102. }
  103. func apacheLogCapture(logfile string) {
  104. if _, err := os.Stat(logfile); err != nil {
  105. log.Fatalf("%s: %s", logfile, err)
  106. }
  107. t, err := tail.TailFile(logfile, tail.Config{
  108. Follow: true, // follow the file
  109. ReOpen: true, // reopen log file when it gets closed/rotated
  110. Logger: tail.DiscardingLogger, // don't log anything
  111. Location: &tail.SeekInfo{Offset: 0, Whence: 2}, // start at the end of the file
  112. })
  113. if err != nil {
  114. log.Fatalf("%s: %s", logfile, err)
  115. }
  116. var p axslogparser.Parser
  117. parserSet := false
  118. for line := range t.Lines {
  119. l := line.Text
  120. if !parserSet {
  121. p, _, err = axslogparser.GuessParser(l)
  122. if err != nil {
  123. log.Println(err)
  124. continue
  125. }
  126. parserSet = true
  127. }
  128. logEntry, err := p.Parse(l)
  129. if err != nil {
  130. log.Println(err)
  131. continue
  132. }
  133. remote := logEntry.Host
  134. if config.UseXForwardedAsSource && logEntry.ForwardedFor != "" {
  135. remote = logEntry.ForwardedFor
  136. }
  137. // only use the first host in case there are multiple hosts in the log
  138. if cidx := strings.Index(remote, ","); cidx >= 0 {
  139. remote = remote[0:cidx]
  140. }
  141. // extract the virtual host
  142. var virtualHost string
  143. vhost := logEntry.VirtualHost
  144. if vhost != "" {
  145. vhostAndPort := strings.Split(logEntry.VirtualHost, ":")
  146. virtualHost = vhostAndPort[0]
  147. if config.UseVhostAsSource {
  148. virtualHost = virtualHost[0:strings.Index(virtualHost, ",")]
  149. vhIP := net.ParseIP(virtualHost)
  150. if vhIP != nil {
  151. remote = virtualHost
  152. }
  153. }
  154. } else {
  155. if config.HostName != "" {
  156. vhost = config.HostName
  157. } else {
  158. vhost = "[not available]"
  159. }
  160. }
  161. request := data.Request{
  162. IpSrc: remote,
  163. IpDst: "127.0.0.1",
  164. PortSrc: 0,
  165. PortDst: 0,
  166. TcpSeq: 0,
  167. CreatedAt: logEntry.Time.UnixNano(),
  168. Url: logEntry.RequestURI,
  169. Method: logEntry.Method,
  170. Host: virtualHost,
  171. Protocol: logEntry.Protocol,
  172. Origin: remote,
  173. Source: remote,
  174. Referer: logEntry.Referer,
  175. XForwardedFor: logEntry.ForwardedFor,
  176. UserAgent: logEntry.UserAgent,
  177. }
  178. if config.Trace {
  179. log.Printf("[%s] %s\n", request.Source, request.Url)
  180. }
  181. count++
  182. publishRequest(config.NatsQueue, &request)
  183. }
  184. }