Add SourceAddr and TLS metadata to lj.Batch

CHANGELOG.md

@@ -6,6 +6,8 @@ This project adheres to [Semantic Versioning](http://semver.org/).
 
 ### Added
 
+- Add connection metadata to `lj.Batch`. [#22](https://github.com/elastic/go-lumber/pull/22)
+
 ### Changed
 
 - Require Go 1.17 to use module. [#28](https://github.com/elastic/go-lumber/pull/28)

cmd/tst-lj/main.go

@@ -21,7 +21,7 @@
 // server supports all lumberjack protocol versions, which must be explicitely enabled
 // from command line. For printing list of known command line flags run:
 //
-//  tst-lj -h
+//	tst-lj -h
 package main
 
 import (
@@ -75,7 +75,7 @@ func main() {
 	}()
 
 	printLog := func(batch *lj.Batch) bool {
-		log.Printf("Received batch of %v events\n", len(batch.Events))
+		log.Printf("Received batch of %v events from %s", len(batch.Events), batch.SourceAddr)
 		return true
 	}
 

lj/lj.go

@@ -18,17 +18,28 @@
 // Package lj implements common lumberjack types and functions.
 package lj
 
+import (
+	"crypto/tls"
+)
+
 // Batch is an ACK-able batch of events as has been received by lumberjack
 // server implemenentations. Batches must be ACKed, for the server
 // implementations returning an ACK to it's clients.
 type Batch struct {
-	Events []interface{}
-	ack    chan struct{}
+	ack        chan struct{}
+	TLS        *tls.ConnectionState // TLS connection metadata. Nil for non-TLS connections.
+	SourceAddr string               // Source address of the connection.
+	Events     []interface{}
 }
 
 // NewBatch creates a new ACK-able batch.
-func NewBatch(evts []interface{}) *Batch {
-	return &Batch{evts, make(chan struct{})}
+func NewBatch(evts []interface{}, remoteAddr string, tlsState *tls.ConnectionState) *Batch {
+	return &Batch{
+		ack:        make(chan struct{}),
+		TLS:        tlsState,
+		SourceAddr: remoteAddr,
+		Events:     evts,
+	}
 }
 
 // ACK acknowledges a batch initiating propagation of ACK to clients.

server/v1/reader.go

@@ -19,6 +19,7 @@ package v1
 
 import (
 	"bufio"
+	"crypto/tls"
 	"encoding/binary"
 	"io"
 	"net"
@@ -32,18 +33,26 @@ import (
 )
 
 type reader struct {
-	in      *bufio.Reader
-	conn    net.Conn
-	timeout time.Duration
-	buf     []byte
+	conn       net.Conn
+	in         *bufio.Reader
+	tlsState   *tls.ConnectionState
+	remoteAddr string
+	buf        []byte
+	timeout    time.Duration
 }
 
 func newReader(c net.Conn, to time.Duration) *reader {
 	r := &reader{
-		in:      bufio.NewReader(c),
-		conn:    c,
-		timeout: to,
-		buf:     make([]byte, 0, 64),
+		conn:       c,
+		in:         bufio.NewReader(c),
+		remoteAddr: c.RemoteAddr().String(),
+		buf:        make([]byte, 0, 64),
+		timeout:    to,
+	}
+
+	if tlsConn, ok := c.(*tls.Conn); ok {
+		s := tlsConn.ConnectionState()
+		r.tlsState = &s
 	}
 	return r
 }
@@ -76,7 +85,7 @@ func (r *reader) ReadBatch() (*lj.Batch, error) {
 		return nil, err
 	}
 
-	return lj.NewBatch(events), nil
+	return lj.NewBatch(events, r.remoteAddr, r.tlsState), nil
 }
 
 func (r *reader) readEvents(in io.Reader, events []interface{}) ([]interface{}, error) {

server/v2/reader.go

@@ -19,6 +19,7 @@ package v2
 
 import (
 	"bufio"
+	"crypto/tls"
 	"encoding/binary"
 	"io"
 	"net"
@@ -32,22 +33,30 @@ import (
 )
 
 type reader struct {
-	in      *bufio.Reader
-	conn    net.Conn
-	timeout time.Duration
-	decoder jsonDecoder
-	buf     []byte
+	conn       net.Conn
+	in         *bufio.Reader
+	tlsState   *tls.ConnectionState
+	decoder    jsonDecoder
+	remoteAddr string
+	buf        []byte
+	timeout    time.Duration
 }
 
 type jsonDecoder func([]byte, interface{}) error
 
 func newReader(c net.Conn, to time.Duration, jsonDecoder jsonDecoder) *reader {
 	r := &reader{
-		in:      bufio.NewReader(c),
-		conn:    c,
-		timeout: to,
-		decoder: jsonDecoder,
-		buf:     make([]byte, 0, 64),
+		conn:       c,
+		in:         bufio.NewReader(c),
+		decoder:    jsonDecoder,
+		remoteAddr: c.RemoteAddr().String(),
+		buf:        make([]byte, 0, 64),
+		timeout:    to,
+	}
+
+	if tlsConn, ok := c.(*tls.Conn); ok {
+		s := tlsConn.ConnectionState()
+		r.tlsState = &s
 	}
 	return r
 }
@@ -80,7 +89,7 @@ func (r *reader) ReadBatch() (*lj.Batch, error) {
 		return nil, err
 	}
 
-	return lj.NewBatch(events), nil
+	return lj.NewBatch(events, r.remoteAddr, r.tlsState), nil
 }
 
 func (r *reader) readEvents(in io.Reader, events []interface{}) ([]interface{}, error) {