mirror_exporter/nginx.go

package main

import (
   "log"
   "net"
   "os"
   "regexp"
   "strconv"
   "strings"
   "time"

   "github.com/hpcloud/tail"
   "github.com/prometheus/client_golang/prometheus"
   "github.com/satyrius/gonx"
)

type NginxExporter struct {
   AccessLogPath string

   responses *prometheus.SummaryVec
   error_responses *prometheus.SummaryVec
}

func NewNginxExporter(accessLogPath string) (*NginxExporter, error) {
   const subsystem = "http"

   return &NginxExporter{
      AccessLogPath: accessLogPath,
      responses: prometheus.NewSummaryVec(
         prometheus.SummaryOpts{
            Namespace: Namespace,
            Subsystem: subsystem,
            Name: "responses",
            Help: "Summary of HTTP responses",
         },
         []string{"project", "network", "protocol"}),
      error_responses: prometheus.NewSummaryVec(
         prometheus.SummaryOpts{
            Namespace: Namespace,
            Subsystem: subsystem,
            Name: "error_responses",
            Help: "Summary of error HTTP responses",
         },
         []string{"project", "network", "protocol"}),
   }, nil
}

// Implements prometheus.Collector
func (e *NginxExporter) Describe(ch chan<- *prometheus.Desc) {
   e.responses.Describe(ch)
   e.error_responses.Describe(ch)
}

// Implements prometheus.Collector
func (e *NginxExporter) Collect(ch chan<- prometheus.Metric) {
   e.responses.Collect(ch)
   e.error_responses.Collect(ch)
}

var nginxProjectRe = regexp.MustCompile("(?i)^\\w+ /(?:pub/)?([^/?]+)/[^\\s]* HTTP")

func (e *NginxExporter) processLogLine(line string) {
   lineReader := strings.NewReader(line)
   reader := gonx.NewReader(
      lineReader,
      `$remote_addr - $remote_user [$time_local] "$request" $status $bytes_sent "$http_referer" "$http_user_agent"`)

   if reader == nil {
      log.Printf("Failed to create reader for line: %s\n", line)
      return
   }

   // Read the entry on the line
   entry, err := reader.Read()
   if err != nil {
      log.Printf("Failed to get entry on line: %s\n", line)
      return
   }

   // Process
   sourceIP, err := entry.Field("remote_addr")
   network := "unknown"
   protocol := "unknown"

   if err == nil {
      ip := net.ParseIP(sourceIP)
      network = IdentifyNetwork(ip)
      protocol = IdentifyIPProtocol(ip)
   } else {
      log.Println(err)
   }

   request, err := entry.Field("request")
   project := "unknown"

   if err == nil {
      match := nginxProjectRe.FindStringSubmatch(request)

      if len(match) > 1 && IsMirroredProject(match[1]) {
         project = match[1]
      } else {
         if len(match) > 1 { log.Println(match[1], "not found") }
         project = "none"
      }
   } else {
      log.Println(err)
   }

   sizeStr, err := entry.Field("bytes_sent")
   if err != nil {
      sizeStr = "0"
   }
   size, _ := strconv.ParseFloat(sizeStr, 64)

   statusStr, err := entry.Field("status")
   success := false

   if err == nil {
      responseCode, _ := strconv.Atoi(statusStr)

      if responseCode >= 100 && responseCode <= 399 {
         success = true
      }
   }

   // Setup labels
   labels := prometheus.Labels{
      "project": project,
      "network": network,
      "protocol": protocol,
   }

   // Increment totals
   e.responses.With(labels).Observe(size)

   if !success {
      e.error_responses.With(labels).Observe(size)
   }
}

func (e *NginxExporter) Monitor() {
   tailConfig := tail.Config{
      Follow: true,
      ReOpen: true,
      Poll: true,
      Location: &tail.SeekInfo{
         Offset: 0,
         Whence: os.SEEK_END,
      },
   }
   t, err := tail.TailFile(e.AccessLogPath, tailConfig)

   // Continue to retry every 5 seconds
   for ; err != nil; {
      t, err = tail.TailFile(e.AccessLogPath, tailConfig)

      if err != nil {
         time.Sleep(time.Second * 5)
         log.Println("Error starting nginx tail (retrying in 5 seconds)", err)
      }
   }

   for line := range t.Lines {
      go e.processLogLine(line.Text)
   }
}