diff --git a/internal/debug/flags.go b/internal/debug/flags.go index 3c85749402..0edd5b0d6d 100644 --- a/internal/debug/flags.go +++ b/internal/debug/flags.go @@ -86,13 +86,40 @@ var ( Name: "trace", Usage: "Write execution trace to the given file", } + splunkurlFlag = cli.StringFlag{ + Name: "splunkurl", + Usage: "URL to Splunk HTTP Event Collector", + } + splunktokenFlag = cli.StringFlag{ + Name: "splunktoken", + Usage: "Splunk HTTP Event Collector token", + } + splunksourceFlag = cli.StringFlag{ + Name: "splunksource", + Usage: "Splunk source field value, description of the source of the event", + Value: "geth", + } + splunkindexFlag = cli.StringFlag{ + Name: "splunkindex", + Usage: "Splunk Index, optional name of the Splunk index to store the event in", + } + splunksourcetypeFlag = cli.StringFlag{ + Name: "splunksourcetype", + Usage: "Splunk source type, optional name of a sourcetype field value", + } + splunkskiptlsverifyFlag = cli.BoolFlag{ + Name: "splunkskiptlsverify", + Usage: "Skip verifying the certificate of the HTTP Event Collector", + } ) // Flags holds all command-line flags required for debugging. var Flags = []cli.Flag{ verbosityFlag, vmoduleFlag, backtraceAtFlag, debugFlag, pprofFlag, pprofAddrFlag, pprofPortFlag, - memprofilerateFlag, blockprofilerateFlag, cpuprofileFlag, traceFlag, + memprofilerateFlag, blockprofilerateFlag, cpuprofileFlag, + traceFlag, splunkurlFlag, splunktokenFlag, splunksourceFlag, + splunkindexFlag, splunksourcetypeFlag, splunkskiptlsverifyFlag, } var ( @@ -114,6 +141,24 @@ func init() { // It should be called as early as possible in the program. func Setup(ctx *cli.Context) error { // logging + splunkURL := ctx.GlobalString(splunkurlFlag.Name) + if splunkURL != "" { + splunkToken := ctx.GlobalString(splunktokenFlag.Name) + splunkSource := ctx.GlobalString(splunksourceFlag.Name) + splunkSourceType := ctx.GlobalString(splunksourcetypeFlag.Name) + splunkIndex := ctx.GlobalString(splunkindexFlag.Name) + splunkSkipTLSVerify := ctx.GlobalBool(splunkskiptlsverifyFlag.Name) + originalHandler := log.NewGlogHandler(ostream) + originalHandler.Verbosity(log.Lvl(ctx.GlobalInt(verbosityFlag.Name))) + originalHandler.Vmodule(ctx.GlobalString(vmoduleFlag.Name)) + originalHandler.BacktraceAt(ctx.GlobalString(backtraceAtFlag.Name)) + splunkstream, err := log.SplunkHandler(splunkURL, splunkToken, splunkSource, splunkSourceType, splunkIndex, + splunkSkipTLSVerify, log.TerminalFormat(false), originalHandler) + if err != nil { + return err + } + glogger.SetHandler(splunkstream) + } log.PrintOrigins(ctx.GlobalBool(debugFlag.Name)) glogger.Verbosity(log.Lvl(ctx.GlobalInt(verbosityFlag.Name))) glogger.Vmodule(ctx.GlobalString(vmoduleFlag.Name)) diff --git a/log/handler.go b/log/handler.go index 3c99114dcb..0a8588b512 100644 --- a/log/handler.go +++ b/log/handler.go @@ -1,13 +1,17 @@ package log import ( + "crypto/tls" "fmt" "io" "net" + "net/http" "os" "reflect" "sync" + "time" + "github.com/ethereum/go-ethereum/log/splunk" "github.com/go-stack/stack" ) @@ -81,6 +85,34 @@ func NetHandler(network, addr string, fmtr Format) (Handler, error) { return closingHandler{conn, StreamHandler(conn, fmtr)}, nil } +// SplunkHandler posts data to Splunk HTTP Event Collector. +func SplunkHandler(url, token, source, sourcetype, index string, skipTLSVerify bool, fmtr Format, + originalHandler Handler) (Handler, error) { + originalLogger := logger{[]interface{}{}, new(swapHandler)} + originalLogger.SetHandler(originalHandler) + tr := &http.Transport{TLSClientConfig: &tls.Config{InsecureSkipVerify: skipTLSVerify}} + httpClient := &http.Client{Timeout: time.Second * 20, Transport: tr} + splunkClient := splunk.NewClient( + httpClient, + url, + token, + source, + sourcetype, + index, + ) + writer := &splunk.Writer{ + Client: splunkClient, + MaxRetries: 2, + } + go func() { + for { + err := <-writer.Errors() + originalLogger.Warn("Error sending info to Splunk", "err", err) + } + }() + return StreamHandler(writer, fmtr), nil +} + // XXX: closingHandler is essentially unused at the moment // it's meant for a future time when the Handler interface supports // a possible Close() operation diff --git a/log/splunk/splunk.go b/log/splunk/splunk.go new file mode 100644 index 0000000000..10d7f2c990 --- /dev/null +++ b/log/splunk/splunk.go @@ -0,0 +1,141 @@ +package splunk + +import ( + "bytes" + "crypto/tls" + "encoding/json" + "errors" + "io" + "io/ioutil" + "net/http" + "os" + "time" +) + +type Event struct { + Time int64 `json:"time"` // epoch time in seconds + Host string `json:"host"` // hostname + Source string `json:"source,omitempty"` // optional description of the source of the event; typically the app's name + SourceType string `json:"sourcetype,omitempty"` // optional name of a Splunk parsing configuration; this is usually inferred by Splunk + Index string `json:"index,omitempty"` // optional name of the Splunk index to store the event in; not required if the token has a default index set in Splunk + Event interface{} `json:"event"` // throw any useful key/val pairs here +} + +type Client struct { + HTTPClient *http.Client + URL string + Hostname string + Token string + Source string + SourceType string + Index string +} + +func NewClient(httpClient *http.Client, URL string, Token string, Source string, SourceType string, Index string) *Client { + if httpClient == nil { + tr := &http.Transport{TLSClientConfig: &tls.Config{InsecureSkipVerify: false}} + httpClient = &http.Client{Timeout: time.Second * 20, Transport: tr} + } + hostname, _ := os.Hostname() + c := &Client{ + HTTPClient: httpClient, + URL: URL, + Hostname: hostname, + Token: Token, + Source: Source, + SourceType: SourceType, + Index: Index, + } + return c +} + +func (c *Client) NewEvent(event interface{}, source string, sourcetype string, index string) *Event { + e := &Event{ + Time: time.Now().Unix(), + Host: c.Hostname, + Source: source, + SourceType: sourcetype, + Index: index, + Event: event, + } + return e +} + +func (c *Client) NewEventWithTime(t int64, event interface{}, source string, sourcetype string, index string) *Event { + e := &Event{ + Time: t, + Host: c.Hostname, + Source: source, + SourceType: sourcetype, + Index: index, + Event: event, + } + return e +} + +func (c *Client) Log(event interface{}) error { + log := c.NewEvent(event, c.Source, c.SourceType, c.Index) + return c.LogEvent(log) +} + +func (c *Client) LogWithTime(t int64, event interface{}) error { + log := c.NewEventWithTime(t, event, c.Source, c.SourceType, c.Index) + return c.LogEvent(log) +} + +func (c *Client) LogEvent(e *Event) error { + b, err := json.Marshal(e) + if err != nil { + return err + } + return c.doRequest(bytes.NewBuffer(b)) +} + +func (c *Client) LogEvents(events []*Event) error { + buf := new(bytes.Buffer) + for _, e := range events { + b, err := json.Marshal(e) + if err != nil { + return err + } + buf.Write(b) + buf.WriteString("\r\n\r\n") + } + return c.doRequest(buf) +} + +func (c *Client) Writer() io.Writer { + return &Writer{ + Client: c, + } +} + +func (c *Client) doRequest(b *bytes.Buffer) error { + url := c.URL + req, err := http.NewRequest("POST", url, b) + if err != nil { + return err + } + req.Header.Add("Content-Type", "application/json") + req.Header.Add("Authorization", "Splunk "+c.Token) + + res, err := c.HTTPClient.Do(req) + if err != nil { + return err + } + + defer res.Body.Close() + + switch res.StatusCode { + case 200: + io.Copy(ioutil.Discard, res.Body) + return nil + default: + buf := new(bytes.Buffer) + buf.ReadFrom(res.Body) + responseBody := buf.String() + err = errors.New(responseBody) + + } + return err +} diff --git a/log/splunk/writer.go b/log/splunk/writer.go new file mode 100644 index 0000000000..ffbc00b370 --- /dev/null +++ b/log/splunk/writer.go @@ -0,0 +1,105 @@ +package splunk + +import ( + "sync" + "time" +) + +const ( + bufferSize = 10000 + defaultInterval = 5 * time.Second + defaultThreshold = 100 + defaultRetries = 0 +) + +type Writer struct { + Client *Client + // How often the write buffer should be flushed to splunk + FlushInterval time.Duration + // How many Write()'s before buffer should be flushed to splunk + FlushThreshold int + // Max number of retries we should do when we flush the buffer + MaxRetries int + dataChan chan *message + errors chan error + once sync.Once +} + +type message struct { + data string + writtenAt time.Time +} + +func (w *Writer) initialize() { + w.once.Do(func() { + w.dataChan = make(chan *message, bufferSize) + w.errors = make(chan error, bufferSize) + go w.listen() + }) +} + +func (w *Writer) Write(b []byte) (int, error) { + w.initialize() + w.dataChan <- &message{ + data: string(b), + writtenAt: time.Now(), + } + return len(b), nil +} + +func (w *Writer) Errors() <-chan error { + w.initialize() + return w.errors +} + +func (w *Writer) listen() { + if w.FlushInterval <= 0 { + w.FlushInterval = defaultInterval + } + if w.FlushThreshold == 0 { + w.FlushThreshold = defaultThreshold + } + if w.MaxRetries == 0 { + w.MaxRetries = defaultRetries + } + ticker := time.NewTicker(w.FlushInterval) + buffer := make([]*message, 0) + + flush := func() { + go w.send(buffer, w.MaxRetries) + buffer = make([]*message, 0) + } + for { + select { + case <-ticker.C: + if len(buffer) > 0 { + flush() + } + case d := <-w.dataChan: + buffer = append(buffer, d) + if len(buffer) > w.FlushThreshold { + flush() + } + } + } +} + +func (w *Writer) send(messages []*message, retries int) { + events := make([]*Event, len(messages)) + for i, m := range messages { + events[i] = w.Client.NewEventWithTime(m.writtenAt.Unix(), m.data, w.Client.Source, w.Client.SourceType, w.Client.Index) + } + err := w.Client.LogEvents(events) + if err != nil { + for i := 0; i < retries; i++ { + err = w.Client.LogEvents(events) + if err == nil { + return + } + } + select { + case w.errors <- err: + default: + } + } +} diff --git a/log/splunk/writer_test.go b/log/splunk/writer_test.go new file mode 100644 index 0000000000..e59fe981fa --- /dev/null +++ b/log/splunk/writer_test.go @@ -0,0 +1,91 @@ +package splunk + +import ( + "fmt" + "io/ioutil" + "net/http" + "net/http/httptest" + "strings" + "sync" + "testing" + "time" +) + +func TestWriter_Write(t *testing.T) { + numWrites := 1000 + numMessages := 0 + lock := sync.Mutex{} + notify := make(chan bool, numWrites) + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + b, _ := ioutil.ReadAll(r.Body) + split := strings.Split(string(b), "\n") + num := 0 + // Since we batch our logs up before we send them: + // Increment our messages counter by one for each JSON object we got in this response + // We don't know how many responses we'll get, we only care about the number of messages + for _, line := range split { + if strings.HasPrefix(line, "{") { + num++ + notify <- true + } + } + lock.Lock() + numMessages = numMessages + num + lock.Unlock() + })) + + // Create a writer that's flushing constantly. We want this test to run + // quickly + writer := Writer{ + Client: NewClient(server.Client(), server.URL, "", "", "", ""), + FlushInterval: 1 * time.Millisecond, + } + // Send a bunch of messages in separate goroutines to make sure we're properly + // testing Writer's concurrency promise + for i := 0; i < numWrites; i++ { + go writer.Write([]byte(fmt.Sprintf("%d", i))) + } + // To notify our test we've collected everything we need. + doneChan := make(chan bool) + go func() { + for i := 0; i < numWrites; i++ { + // Do nothing, just loop through to the next one + <-notify + } + doneChan <- true + }() + select { + case <-doneChan: + // Do nothing, we're good + case <-time.After(1 * time.Second): + t.Errorf("Timed out waiting for messages") + } + // We may have received more than numWrites amount of messages, check that case + if numMessages != numWrites { + t.Errorf("Didn't get the right number of messages, expected %d, got %d", numWrites, numMessages) + } +} + +func TestWriter_Errors(t *testing.T) { + numMessages := 1000 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusBadRequest) + fmt.Fprintln(w, "bad request") + })) + writer := Writer{ + Client: NewClient(server.Client(), server.URL, "", "", "", ""), + // Will flush after the last message is sent + FlushThreshold: numMessages - 1, + // Don't let the flush interval cause raciness + FlushInterval: 5 * time.Minute, + } + for i := 0; i < numMessages; i++ { + _, _ = writer.Write([]byte("some data")) + } + select { + case <-writer.Errors(): + // good to go, got our error + case <-time.After(1 * time.Second): + t.Errorf("Timed out waiting for error, should have gotten 1 error") + } +}