log,internal/debug: Add Splunk logging capability

Adds the ability to log directly to Splunk using the HTTP Event
Collector (HEC).

Some of the code is taken after
github.com/ZachtimusPrime/Go-Splunk-HTTP, but has been patched and
edited to support this use case best.

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))

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

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
+}

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:
+		}
+	}
+}

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")
+	}
+}