package graylog import ( "bytes" "compress/zlib" "crypto/rand" "encoding/binary" ejson "encoding/json" "fmt" "github.com/influxdata/telegraf" "github.com/influxdata/telegraf/plugins/outputs" "io" "math" "net" "os" ) const ( defaultGraylogEndpoint = "127.0.0.1:12201" defaultConnection = "wan" defaultMaxChunkSizeWan = 1420 defaultMaxChunkSizeLan = 8154 ) type GelfConfig struct { GraylogEndpoint string Connection string MaxChunkSizeWan int MaxChunkSizeLan int } type Gelf struct { GelfConfig } func NewGelfWriter(config GelfConfig) *Gelf { if config.GraylogEndpoint == "" { config.GraylogEndpoint = defaultGraylogEndpoint } if config.Connection == "" { config.Connection = defaultConnection } if config.MaxChunkSizeWan == 0 { config.MaxChunkSizeWan = defaultMaxChunkSizeWan } if config.MaxChunkSizeLan == 0 { config.MaxChunkSizeLan = defaultMaxChunkSizeLan } g := &Gelf{GelfConfig: config} return g } func (g *Gelf) Write(message []byte) (n int, err error) { compressed := g.compress(message) chunksize := g.GelfConfig.MaxChunkSizeWan length := compressed.Len() if length > chunksize { chunkCountInt := int(math.Ceil(float64(length) / float64(chunksize))) id := make([]byte, 8) rand.Read(id) for i, index := 0, 0; i < length; i, index = i+chunksize, index+1 { packet := g.createChunkedMessage(index, chunkCountInt, id, &compressed) _, err = g.send(packet.Bytes()) if err != nil { return 0, err } } } else { _, err = g.send(compressed.Bytes()) if err != nil { return 0, err } } n = len(message) return } func (g *Gelf) createChunkedMessage(index int, chunkCountInt int, id []byte, compressed *bytes.Buffer) bytes.Buffer { var packet bytes.Buffer chunksize := g.getChunksize() packet.Write(g.intToBytes(30)) packet.Write(g.intToBytes(15)) packet.Write(id) packet.Write(g.intToBytes(index)) packet.Write(g.intToBytes(chunkCountInt)) packet.Write(compressed.Next(chunksize)) return packet } func (g *Gelf) getChunksize() int { if g.GelfConfig.Connection == "wan" { return g.GelfConfig.MaxChunkSizeWan } if g.GelfConfig.Connection == "lan" { return g.GelfConfig.MaxChunkSizeLan } return g.GelfConfig.MaxChunkSizeWan } func (g *Gelf) intToBytes(i int) []byte { buf := new(bytes.Buffer) binary.Write(buf, binary.LittleEndian, int8(i)) return buf.Bytes() } func (g *Gelf) compress(b []byte) bytes.Buffer { var buf bytes.Buffer comp := zlib.NewWriter(&buf) comp.Write(b) comp.Close() return buf } func (g *Gelf) send(b []byte) (n int, err error) { udpAddr, err := net.ResolveUDPAddr("udp", g.GelfConfig.GraylogEndpoint) if err != nil { return } conn, err := net.DialUDP("udp", nil, udpAddr) if err != nil { return } n, err = conn.Write(b) return } type Graylog struct { Servers []string writer io.Writer } var sampleConfig = ` ## Udp endpoint for your graylog instance. servers = ["127.0.0.1:12201", "192.168.1.1:12201"] ` func (g *Graylog) Connect() error { writers := []io.Writer{} if len(g.Servers) == 0 { g.Servers = append(g.Servers, "localhost:12201") } for _, server := range g.Servers { w := NewGelfWriter(GelfConfig{GraylogEndpoint: server}) writers = append(writers, w) } g.writer = io.MultiWriter(writers...) return nil } func (g *Graylog) Close() error { return nil } func (g *Graylog) SampleConfig() string { return sampleConfig } func (g *Graylog) Description() string { return "Send telegraf metrics to graylog(s)" } func (g *Graylog) Write(metrics []telegraf.Metric) error { if len(metrics) == 0 { return nil } for _, metric := range metrics { values, err := serialize(metric) if err != nil { return err } for _, value := range values { _, err := g.writer.Write([]byte(value)) if err != nil { return fmt.Errorf("FAILED to write message: %s, %s", value, err) } } } return nil } func serialize(metric telegraf.Metric) ([]string, error) { out := []string{} m := make(map[string]interface{}) m["version"] = "1.1" m["timestamp"] = metric.UnixNano() / 1000000000 m["short_message"] = " " m["name"] = metric.Name() if host, ok := metric.Tags()["host"]; ok { m["host"] = host } else { host, err := os.Hostname() if err != nil { return []string{}, err } m["host"] = host } for key, value := range metric.Fields() { nkey := fmt.Sprintf("_%s", key) m[nkey] = value } serialized, err := ejson.Marshal(m) if err != nil { return []string{}, err } out = append(out, string(serialized)) return out, nil } func init() { outputs.Add("graylog", func() telegraf.Output { return &Graylog{} }) }