Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Add simple Logstash adapter #64

Closed
wants to merge 6 commits into from
Closed
Show file tree
Hide file tree
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
63 changes: 63 additions & 0 deletions adapters/logstash/logstash.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,63 @@
package logstash

import (
"encoding/json"
"errors"
"io"
"net"
"os"

"github.com/gliderlabs/logspout/router"
)

var hostname string

func init() {
router.AdapterFactories.Register(NewLogstashAdapter, "logstash")

hostname, _ = os.Hostname()
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

What's this?

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I see, it's a global. Maybe set this up as a field of the adapter object and do this in its constructor.

}

// LogstashAdapter is an adapter that streams UPD JSON to Logstash.
type LogstashAdapter struct {
conn net.Conn
route *router.Route
}

// NewLogstashAdapter creates a LogstashAdapter with UDP transport.
func NewLogstashAdapter(route *router.Route) (router.LogAdapter, error) {
transport, found := router.AdapterTransports.Lookup(route.AdapterTransport("udp"))
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is actually saying "udp" is the default transport (by using route.AdapterTransport), but allows any other supported transport if specified ("logstash+tcp://"). Does logstash support other transports or should this be hardcoded to udp?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I believe TCP should work also, but I have not tried it.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Okay. Your comments suggest otherwise, so maybe consider removing specific mention of UDP in them.

if !found {
return nil, errors.New("unable to find adapter: " + route.Adapter)
}

conn, err := transport.Dial(route.Address, route.Options)
if err != nil {
return nil, err
}

return &LogstashAdapter{
route: route,
conn: conn,
}, nil
}

func (adapter *LogstashAdapter) Stream(logstream chan *router.Message) {
for m := range logstream {
msg := LogstashMessage{
Time: m.Time.Unix(),
Message: m.Data,
Hostname: hostname,
Image: m.Container.Config.Image,
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'm guessing logstash just supports arbitrary JSON keys? If so, don't you think some people might want more than just container image? Do you have ideas on how to make this more configurable?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

That is true. At this point we are doing further parsing of the message (which is in logfmt format by logrus) in the filter section of logstash. Maybe the container ID and some other Docker metadata could be good to have as keys.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It seems like this adapter is not as general as you think. Maybe it deserves to live as your own third-party module?

I say this because if I were to do a general Logstash adapter, I probably wouldn't use JSON+UDP which is just one possible input format ... If I were to pick a canonical Logstash adapter machanism, I'd use the logstash-forwarder protocol, perhaps even use it directly as a library.

Don't you agree?

}
js, _ := json.Marshal(msg)
io.WriteString(adapter.conn, string(js))
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

You don't need to use io.WriteString when you're converting a byte array to a string... you can just say adapter.conn.Write(js).. be sure to error check though.

}
}

type LogstashMessage struct {
Time int64 `json:"time"`
Message string `json:"message"`
Hostname string `json:"hostname"`
Image string `json:"image"`
}
1 change: 1 addition & 0 deletions modules.go
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
package main

import (
_ "github.com/gliderlabs/logspout/adapters/logstash"
_ "github.com/gliderlabs/logspout/adapters/raw"
_ "github.com/gliderlabs/logspout/adapters/syslog"
_ "github.com/gliderlabs/logspout/httpstream"
Expand Down