PDD measurement tool

asterisk.conf

exten => _XXXXXX.,n,Set(SessionId=${UNIQUEID})
exten => _XXXXXX.,n,Set(CDR(my_uuid)=${SessionId})
exten => _XXXXXX.,n,SIPaddheader(X-CallId: ${SessionId})

main.go

package main
import (
  "code.google.com/p/goconf/conf"
  "bitbucket.org/sdr/sip_parser/src"
  "github.com/akrennmair/gopcap"
  "github.com/fzzy/radix/redis"
  "time"
  "log/syslog"
  "fmt"
)


type PddCall struct {
  XCallId string
  StartedAt time.Time
}

type PddReport struct {
  CallId string
  Pdd float64
}

func (ch *PddCall) duration() time.Duration {
  return time.Now().Sub(ch.StartedAt)
}

var slog *syslog.Writer
var err error
var redis_report_client *redis.Client
var reports_chan chan PddReport
var redis_uri string

func main() {

  calls := make(map[string]PddCall)
  reports_chan = make(chan PddReport, 100)

  slog, err = syslog.New(syslog.LOG_INFO, "[pdd-watcher]") 
  if err != nil { 
    fmt.Printf("%s", err) 
  } 



  config, err := conf.ReadConfigFile("/etc/pdd-watcher.conf")
  checkError(err)
  redis_uri, err = config.GetString("default", "redis_url")
  checkError(err)

  filter, err := config.GetString("default", "filter")
  checkError(err)

  device, err := config.GetString("default", "interface")
  checkError(err)

  slog.Info("Config loaded")
  slog.Info("Redis at " + redis_uri)

  redis_report_client, err = redis.Dial("tcp", redis_uri)
  checkError(err)

  slog.Info("Redis connected")

  go redisReporterWorker(reports_chan)

  redis_report_client.Cmd("SELECT", 0)

  var h *pcap.Pcap
  h, err = pcap.Openlive(device, 65535, false, 0)
  checkError(err)
  defer h.Close()

  err = h.Setfilter(filter)
  checkError(err)

  var x_callid string
  var offset uint32
  offset = 8 + 28 + 6

  for pkt := h.Next(); pkt != nil; pkt = h.Next() {

    raw_data := string(pkt.Data[offset:pkt.Caplen])
    sip_packet := sipparser.ParseMsg(raw_data)

    if sip_packet.Error == nil {

      //We -> Trunk (invite)
      if(sip_packet.StartLine.Method == "INVITE") {
        //Now check for our header
        headers := sip_packet.Headers
        x_callid = ""

        for hdr := range headers {
          if sip_packet.Headers[hdr].Header == "x-callid" {
            x_callid = sip_packet.Headers[hdr].Val
            break
          }
        }

        if x_callid != "" {
          calls[sip_packet.CallId] = PddCall{XCallId: x_callid, StartedAt: time.Now()}
        }
      }

      if(sip_packet.StartLine.Type == sipparser.SIP_RESPONSE) {
        callid := sip_packet.CallId
        code := sip_packet.StartLine.Resp

        call, exists := calls[callid]

        if exists {
          if (code == "100") {
            //Do nothing
          } else {//if (code == "183") || (code == "200") {
            duration := call.duration()
            slog.Info(fmt.Sprintf("PDD For %s is %f\n", call.XCallId, duration.Seconds()))
            reports_chan <- PddReport{CallId: call.XCallId, Pdd: duration.Seconds()}
            delete(calls, callid)
          } //else {
            //delete(calls, callid)
          //}
        }

      }
    }
  }

}

func redisReporterWorker(reports <- chan PddReport) {
  for report := range reports {
    res := redis_report_client.Cmd("SET", "cdr-" + report.CallId, report.Pdd)
    if(res.Err != nil) {
      slog.Info(fmt.Sprintf("Redis connection died. Last error: %q", res.Err))
      redis_report_client, err = redis.Dial("tcp", redis_uri)
      checkError(err)
      reports_chan <- report
    } else {
      redis_report_client.Cmd("EXPIRE", "cdr-" + report.CallId, 10 * 60 * 60)
    }
  }
}

func checkError(err error) {
  if err != nil {
    slog.Err(fmt.Sprintf("%q", err))
  }
}

pdd-watcher.conf

[default]
redis_url=127.0.0.1:6379
filter=port 5060
interface=eth0