scribe_server.go

package main

import (
    "flag"
    "fmt"
    "net"
    "net/rpc"
    "strings"
    "time"

    "github.com/amir/raidman"
    "github.com/samuel/go-thrift/examples/scribe"
    "github.com/samuel/go-thrift/thrift"
)

// implementation

type scribeServiceImplementation struct {
    riemann string
    ttl     float32
}

func report(riemann string, event *raidman.Event) {
    c, err := raidman.Dial("udp", riemann)
    if err != nil {
        panic(err)
    }

    err = c.Send(event)
    if err != nil {
        panic(err)
    }

    c.Close()
}

func (s *scribeServiceImplementation) Log(messages []*scribe.LogEntry) (scribe.ResultCode, error) {
    const format = "2006-01-02 15:04:05"
    const count = 1

    for _, m := range messages {
        var when, err = time.Parse(format, m.Message[:19])
        if err != nil {
            fmt.Printf("ERROR: %+v\n", err)
        }
        var words = strings.Split(m.Message, " ")
        var msg = strings.TrimRight(strings.Join(words[7:], " "), "\n")

        var event = raidman.Event{
            State:   words[3],
            Host:    words[2],
            Service: m.Category,
            Metric:  count,
            Ttl:     s.ttl,
        }

        report(s.riemann, &event)
        fmt.Printf("MSG: %+s %s %s %s %s\n", m.Category, when, words[2], words[3], msg)
    }
    return scribe.ResultCodeOk, nil
}

func main() {
    var riemann = flag.String("riemann", "127.0.0.1:5555", "riemann server to report")
    var port = flag.String("listen", ":1463", "address to listen as scribe")
    var ttl = flag.Float64("ttl", 60.0, "TTL to store a value")
    flag.Parse()

    scribeService := new(scribeServiceImplementation)
    scribeService.riemann = *riemann
    scribeService.ttl = float32(*ttl)
    rpc.RegisterName("Thrift", &scribe.ScribeServer{scribeService})

    ln, err := net.Listen("tcp", *port)
    if err != nil {
        panic(err)
    }
    for {
        conn, err := ln.Accept()
        if err != nil {
            fmt.Printf("ERROR: %+v\n", err)
            continue
        }
        fmt.Printf("New connection %+v\n", conn)
        go rpc.ServeCodec(thrift.NewServerCodec(thrift.NewFramedReadWriteCloser(conn, 0), thrift.NewBinaryProtocol(true, false)))
    }
}