CL-Softwares
/
telegraf
8
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

fix: Graylog plugin TLS support and message format (#9862)

This commit is contained in:
alespour 2021-10-21 16:39:24 +02:00 committed by GitHub
parent c669874750
commit e685f3be46
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
4 changed files with 253 additions and 119 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
etc/telegraf.conf
View File

@ -769,6 +769,13 @@
# ## "telegraf" will be used. # ## "telegraf" will be used.
# ## example: short_message_field = "message" # ## example: short_message_field = "message"
# # short_message_field = "" # # short_message_field = ""
#
# ## Optional TLS Config
# # tls_ca = "/etc/telegraf/ca.pem"
# # tls_cert = "/etc/telegraf/cert.pem"
# # tls_key = "/etc/telegraf/key.pem"
# ## Use TLS but skip chain & host verification
# # insecure_skip_verify = false
# # Configurable HTTP health check resource based on metrics # # Configurable HTTP health check resource based on metrics

14
plugins/outputs/graylog/README.md
View File

@ -18,7 +18,19 @@ This plugin writes to a Graylog instance using the "[GELF][]" format.
## "telegraf" will be used. ## "telegraf" will be used.
## example: short_message_field = "message" ## example: short_message_field = "message"
# short_message_field = "" # short_message_field = ""
## According to GELF payload specification, additional fields names must be prefixed
## with an underscore. Previous versions did not prefix custom field 'name' with underscore.
## Set to true for backward compatibility.
# name_field_no_prefix = false
## Optional TLS Config
# tls_ca = "/etc/telegraf/ca.pem"
# tls_cert = "/etc/telegraf/cert.pem"
# tls_key = "/etc/telegraf/key.pem"
## Use TLS but skip chain & host verification
# insecure_skip_verify = false
``` ```
Server endpoint may be specified without UDP or TCP scheme (eg. "127.0.0.1:12201"). Server endpoint may be specified without UDP or TCP scheme (eg. "127.0.0.1:12201").
In such case, UDP protocol is assumed. In such case, UDP protocol is assumed. TLS config is ignored for UDP endpoints.

85
plugins/outputs/graylog/graylog.go
View File

@ -4,6 +4,7 @@ import (
"bytes" "bytes"
"compress/zlib" "compress/zlib"
"crypto/rand" "crypto/rand"
"crypto/tls"
"encoding/binary" "encoding/binary"
ejson "encoding/json" ejson "encoding/json"
"fmt" "fmt"
@ -16,11 +17,12 @@ import (
"github.com/influxdata/telegraf" "github.com/influxdata/telegraf"
"github.com/influxdata/telegraf/config" "github.com/influxdata/telegraf/config"
tlsint "github.com/influxdata/telegraf/plugins/common/tls"
"github.com/influxdata/telegraf/plugins/outputs" "github.com/influxdata/telegraf/plugins/outputs"
) )
const ( const (
defaultGraylogEndpoint = "127.0.0.1:12201" defaultEndpoint = "127.0.0.1:12201"
defaultConnection = "wan" defaultConnection = "wan"
defaultMaxChunkSizeWan = 1420 defaultMaxChunkSizeWan = 1420
defaultMaxChunkSizeLan = 8154 defaultMaxChunkSizeLan = 8154
@ -29,7 +31,7 @@ const (
) )
type gelfConfig struct { type gelfConfig struct {
GraylogEndpoint string Endpoint string
Connection string Connection string
MaxChunkSizeWan int MaxChunkSizeWan int
MaxChunkSizeLan int MaxChunkSizeLan int
@ -37,6 +39,7 @@ type gelfConfig struct {
type gelf interface { type gelf interface {
io.WriteCloser io.WriteCloser
Connect() error
} }
type gelfCommon struct { type gelfCommon struct {
@ -51,11 +54,12 @@ type gelfUDP struct {
type gelfTCP struct { type gelfTCP struct {
gelfCommon gelfCommon
tlsConfig *tls.Config
} }
func newGelfWriter(cfg gelfConfig, dialer *net.Dialer) gelf { func newGelfWriter(cfg gelfConfig, dialer *net.Dialer, tlsConfig *tls.Config) gelf {
if cfg.GraylogEndpoint == "" { if cfg.Endpoint == "" {
cfg.GraylogEndpoint = defaultGraylogEndpoint cfg.Endpoint = defaultEndpoint
} }
if cfg.Connection == "" { if cfg.Connection == "" {
@ -71,10 +75,10 @@ func newGelfWriter(cfg gelfConfig, dialer *net.Dialer) gelf {
} }
scheme := defaultScheme scheme := defaultScheme
parts := strings.SplitN(cfg.GraylogEndpoint, "://", 2) parts := strings.SplitN(cfg.Endpoint, "://", 2)
if len(parts) == 2 { if len(parts) == 2 {
scheme = strings.ToLower(parts[0]) scheme = strings.ToLower(parts[0])
cfg.GraylogEndpoint = parts[1] cfg.Endpoint = parts[1]
} }
common := gelfCommon{ common := gelfCommon{
gelfConfig: cfg, gelfConfig: cfg,
@ -84,7 +88,7 @@ func newGelfWriter(cfg gelfConfig, dialer *net.Dialer) gelf {
var g gelf var g gelf
switch scheme { switch scheme {
case "tcp": case "tcp":
g = &gelfTCP{gelfCommon: common} g = &gelfTCP{gelfCommon: common, tlsConfig: tlsConfig}
default: default:
g = &gelfUDP{gelfCommon: common} g = &gelfUDP{gelfCommon: common}
} }
@ -178,13 +182,21 @@ func (g *gelfUDP) compress(b []byte) bytes.Buffer {
return buf return buf
} }
func (g *gelfUDP) Connect() error {
conn, err := g.dialer.Dial("udp", g.gelfConfig.Endpoint)
if err != nil {
return err
}
g.conn = conn
return nil
}
func (g *gelfUDP) send(b []byte) error { func (g *gelfUDP) send(b []byte) error {
if g.conn == nil { if g.conn == nil {
conn, err := g.dialer.Dial("udp", g.gelfConfig.GraylogEndpoint) err := g.Connect()
if err != nil { if err != nil {
return err return err
} }
g.conn = conn
} }
_, err := g.conn.Write(b) _, err := g.conn.Write(b)
@ -216,13 +228,27 @@ func (g *gelfTCP) Close() (err error) {
return err return err
} }
func (g *gelfTCP) Connect() error {
var err error
var conn net.Conn
if g.tlsConfig == nil {
conn, err = g.dialer.Dial("tcp", g.gelfConfig.Endpoint)
} else {
conn, err = tls.DialWithDialer(g.dialer, "tcp", g.gelfConfig.Endpoint, g.tlsConfig)
}
if err != nil {
return err
}
g.conn = conn
return nil
}
func (g *gelfTCP) send(b []byte) error { func (g *gelfTCP) send(b []byte) error {
if g.conn == nil { if g.conn == nil {
conn, err := g.dialer.Dial("tcp", g.gelfConfig.GraylogEndpoint) err := g.Connect()
if err != nil { if err != nil {
return err return err
} }
g.conn = conn
} }
_, err := g.conn.Write(b) _, err := g.conn.Write(b)
@ -243,7 +269,9 @@ func (g *gelfTCP) send(b []byte) error {
type Graylog struct { type Graylog struct {
Servers []string `toml:"servers"` Servers []string `toml:"servers"`
ShortMessageField string `toml:"short_message_field"` ShortMessageField string `toml:"short_message_field"`
NameFieldNoPrefix bool `toml:"name_field_noprefix"`
Timeout config.Duration `toml:"timeout"` Timeout config.Duration `toml:"timeout"`
tlsint.ClientConfig
writer io.Writer writer io.Writer
closers []io.WriteCloser closers []io.WriteCloser
@ -260,18 +288,39 @@ var sampleConfig = `
## "telegraf" will be used. ## "telegraf" will be used.
## example: short_message_field = "message" ## example: short_message_field = "message"
# short_message_field = "" # short_message_field = ""
## According to GELF payload specification, additional fields names must be prefixed
## with an underscore. Previous versions did not prefix custom field 'name' with underscore.
## Set to true for backward compatibility.
# name_field_no_prefix = false
## Optional TLS Config
# tls_ca = "/etc/telegraf/ca.pem"
# tls_cert = "/etc/telegraf/cert.pem"
# tls_key = "/etc/telegraf/key.pem"
## Use TLS but skip chain & host verification
# insecure_skip_verify = false
` `
func (g *Graylog) Connect() error { func (g *Graylog) Connect() error {
writers := []io.Writer{} var writers []io.Writer
dialer := net.Dialer{Timeout: time.Duration(g.Timeout)} dialer := &net.Dialer{Timeout: time.Duration(g.Timeout)}
if len(g.Servers) == 0 { if len(g.Servers) == 0 {
g.Servers = append(g.Servers, "localhost:12201") g.Servers = append(g.Servers, "localhost:12201")
} }
tlsCfg, err := g.ClientConfig.TLSConfig()
if err != nil {
return err
}
for _, server := range g.Servers { for _, server := range g.Servers {
w := newGelfWriter(gelfConfig{GraylogEndpoint: server}, &dialer) w := newGelfWriter(gelfConfig{Endpoint: server}, dialer, tlsCfg)
err := w.Connect()
if err != nil {
return fmt.Errorf("failed to connect to server [%s]: %v", server, err)
}
writers = append(writers, w) writers = append(writers, w)
g.closers = append(g.closers, w) g.closers = append(g.closers, w)
} }
@ -319,7 +368,11 @@ func (g *Graylog) serialize(metric telegraf.Metric) ([]string, error) {
m["version"] = "1.1" m["version"] = "1.1"
m["timestamp"] = float64(metric.Time().UnixNano()) / 1_000_000_000 m["timestamp"] = float64(metric.Time().UnixNano()) / 1_000_000_000
m["short_message"] = "telegraf" m["short_message"] = "telegraf"
m["name"] = metric.Name() if g.NameFieldNoPrefix {
m["name"] = metric.Name()
} else {
m["_name"] = metric.Name()
}
if host, ok := metric.GetTag("host"); ok { if host, ok := metric.GetTag("host"); ok {
m["host"] = host m["host"] = host

266
plugins/outputs/graylog/graylog_test.go
View File

@ -3,125 +3,174 @@ package graylog
import ( import (
"bytes" "bytes"
"compress/zlib" "compress/zlib"
"crypto/tls"
"encoding/json" "encoding/json"
"io" "io"
"net" "net"
"sync" "sync"
"testing" "testing"
"time"
tlsint "github.com/influxdata/telegraf/plugins/common/tls"
"github.com/influxdata/telegraf/testutil" "github.com/influxdata/telegraf/testutil"
"github.com/stretchr/testify/assert" "github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require" "github.com/stretchr/testify/require"
) )
func TestWriteDefault(t *testing.T) {
if testing.Short() {
t.Skip("skipping test in short mode.")
}
scenarioUDP(t, "127.0.0.1:12201")
}
func TestWriteUDP(t *testing.T) { func TestWriteUDP(t *testing.T) {
if testing.Short() { tests := []struct {
t.Skip("skipping test in short mode.") name string
instance Graylog
}{
{
name: "default without scheme",
instance: Graylog{
Servers: []string{"127.0.0.1:12201"},
},
},
{
name: "UDP",
instance: Graylog{
Servers: []string{"udp://127.0.0.1:12201"},
},
},
{
name: "UDP non-standard name field",
instance: Graylog{
Servers: []string{"udp://127.0.0.1:12201"},
NameFieldNoPrefix: true,
},
},
} }
scenarioUDP(t, "udp://127.0.0.1:12201") for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
var wg sync.WaitGroup
var wg2 sync.WaitGroup
wg.Add(1)
wg2.Add(1)
go UDPServer(t, &wg, &wg2, &tt.instance)
wg2.Wait()
i := tt.instance
err := i.Connect()
require.NoError(t, err)
defer i.Close()
defer wg.Wait()
metrics := testutil.MockMetrics()
// UDP scenario:
// 4 messages are send
err = i.Write(metrics)
require.NoError(t, err)
err = i.Write(metrics)
require.NoError(t, err)
err = i.Write(metrics)
require.NoError(t, err)
err = i.Write(metrics)
require.NoError(t, err)
})
}
} }
func TestWriteTCP(t *testing.T) { func TestWriteTCP(t *testing.T) {
if testing.Short() { pki := testutil.NewPKI("../../../testutil/pki")
t.Skip("skipping test in short mode.") tlsClientConfig := pki.TLSClientConfig()
tlsServerConfig, err := pki.TLSServerConfig().TLSConfig()
require.NoError(t, err)
tests := []struct {
name string
instance Graylog
tlsServerConfig *tls.Config
}{
{
name: "TCP",
instance: Graylog{
Servers: []string{"tcp://127.0.0.1:12201"},
},
},
{
name: "TLS",
instance: Graylog{
Servers: []string{"tcp://127.0.0.1:12201"},
ClientConfig: tlsint.ClientConfig{
ServerName: "localhost",
TLSCA: tlsClientConfig.TLSCA,
TLSKey: tlsClientConfig.TLSKey,
TLSCert: tlsClientConfig.TLSCert,
},
},
tlsServerConfig: tlsServerConfig,
},
{
name: "TLS no validation",
instance: Graylog{
Servers: []string{"tcp://127.0.0.1:12201"},
ClientConfig: tlsint.ClientConfig{
InsecureSkipVerify: true,
ServerName: "localhost",
TLSKey: tlsClientConfig.TLSKey,
TLSCert: tlsClientConfig.TLSCert,
},
},
tlsServerConfig: tlsServerConfig,
},
} }
scenarioTCP(t, "tcp://127.0.0.1:12201") for _, tt := range tests {
} t.Run(tt.name, func(t *testing.T) {
var wg sync.WaitGroup
var wg2 sync.WaitGroup
var wg3 sync.WaitGroup
wg.Add(1)
wg2.Add(1)
wg3.Add(1)
go TCPServer(t, &wg, &wg2, &wg3, tt.tlsServerConfig)
wg2.Wait()
func scenarioUDP(t *testing.T, server string) { i := tt.instance
var wg sync.WaitGroup err = i.Connect()
var wg2 sync.WaitGroup require.NoError(t, err)
wg.Add(1) defer i.Close()
wg2.Add(1) defer wg.Wait()
go UDPServer(t, &wg, &wg2)
wg2.Wait()
i := Graylog{ metrics := testutil.MockMetrics()
Servers: []string{server},
// TCP scenario:
// 4 messages are send
// -> connection gets forcefully broken after the 2nd message (server closes connection)
// -> the 3rd write fails with error
// -> during the 4th write connection is restored and write is successful
err = i.Write(metrics)
require.NoError(t, err)
err = i.Write(metrics)
require.NoError(t, err)
wg3.Wait()
err = i.Write(metrics)
require.Error(t, err)
err = i.Write(metrics)
require.NoError(t, err)
})
} }
err := i.Connect()
require.NoError(t, err)
metrics := testutil.MockMetrics()
// UDP scenario:
// 4 messages are send
err = i.Write(metrics)
require.NoError(t, err)
err = i.Write(metrics)
require.NoError(t, err)
err = i.Write(metrics)
require.NoError(t, err)
err = i.Write(metrics)
require.NoError(t, err)
wg.Wait()
i.Close()
}
func scenarioTCP(t *testing.T, server string) {
var wg sync.WaitGroup
var wg2 sync.WaitGroup
var wg3 sync.WaitGroup
wg.Add(1)
wg2.Add(1)
wg3.Add(1)
go TCPServer(t, &wg, &wg2, &wg3)
wg2.Wait()
i := Graylog{
Servers: []string{server},
}
err := i.Connect()
require.NoError(t, err)
metrics := testutil.MockMetrics()
// TCP scenario:
// 4 messages are send
// -> connection gets broken after the 2nd message (server closes connection)
// -> the 3rd write ends with error
// -> in the 4th write connection is restored and write is successful
err = i.Write(metrics)
require.NoError(t, err)
err = i.Write(metrics)
require.NoError(t, err)
wg3.Wait()
err = i.Write(metrics)
require.Error(t, err)
err = i.Write(metrics)
require.NoError(t, err)
wg.Wait()
i.Close()
} }
type GelfObject map[string]interface{} type GelfObject map[string]interface{}
func UDPServer(t *testing.T, wg *sync.WaitGroup, wg2 *sync.WaitGroup) { func UDPServer(t *testing.T, wg *sync.WaitGroup, wg2 *sync.WaitGroup, config *Graylog) {
serverAddr, err := net.ResolveUDPAddr("udp", "127.0.0.1:12201") serverAddr, err := net.ResolveUDPAddr("udp", "127.0.0.1:12201")
require.NoError(t, err) require.NoError(t, err)
udpServer, err := net.ListenUDP("udp", serverAddr) udpServer, err := net.ListenUDP("udp", serverAddr)
require.NoError(t, err) require.NoError(t, err)
defer udpServer.Close() defer udpServer.Close()
defer wg.Done() defer wg.Done()
bufR := make([]byte, 1024)
wg2.Done() wg2.Done()
recv := func() { recv := func() {
bufR := make([]byte, 1024)
n, _, err := udpServer.ReadFromUDP(bufR) n, _, err := udpServer.ReadFromUDP(bufR)
require.NoError(t, err) require.NoError(t, err)
@ -135,6 +184,13 @@ func UDPServer(t *testing.T, wg *sync.WaitGroup, wg2 *sync.WaitGroup) {
var obj GelfObject var obj GelfObject
_ = json.Unmarshal(bufW.Bytes(), &obj) _ = json.Unmarshal(bufW.Bytes(), &obj)
require.NoError(t, err) require.NoError(t, err)
assert.Equal(t, obj["short_message"], "telegraf")
if config.NameFieldNoPrefix {
assert.Equal(t, obj["name"], "test1")
} else {
assert.Equal(t, obj["_name"], "test1")
}
assert.Equal(t, obj["_tag1"], "value1")
assert.Equal(t, obj["_value"], float64(1)) assert.Equal(t, obj["_value"], float64(1))
} }
@ -146,29 +202,29 @@ func UDPServer(t *testing.T, wg *sync.WaitGroup, wg2 *sync.WaitGroup) {
recv() recv()
} }
func TCPServer(t *testing.T, wg *sync.WaitGroup, wg2 *sync.WaitGroup, wg3 *sync.WaitGroup) { func TCPServer(t *testing.T, wg *sync.WaitGroup, wg2 *sync.WaitGroup, wg3 *sync.WaitGroup, tlsConfig *tls.Config) {
serverAddr, err := net.ResolveTCPAddr("tcp", "127.0.0.1:12201") tcpServer, err := net.Listen("tcp", "127.0.0.1:12201")
require.NoError(t, err)
tcpServer, err := net.ListenTCP("tcp", serverAddr)
require.NoError(t, err) require.NoError(t, err)
defer tcpServer.Close() defer tcpServer.Close()
defer wg.Done() defer wg.Done()
bufR := make([]byte, 1)
bufW := bytes.NewBuffer(nil)
wg2.Done() wg2.Done()
accept := func() *net.TCPConn { accept := func() net.Conn {
conn, err := tcpServer.AcceptTCP() conn, err := tcpServer.Accept()
require.NoError(t, err) require.NoError(t, err)
_ = conn.SetLinger(0) if tcpConn, ok := conn.(*net.TCPConn); ok {
_ = tcpConn.SetLinger(0)
}
_ = conn.SetDeadline(time.Now().Add(15 * time.Second))
if tlsConfig != nil {
conn = tls.Server(conn, tlsConfig)
}
return conn return conn
} }
conn := accept()
defer conn.Close()
recv := func() { recv := func(conn net.Conn) {
bufW.Reset() bufR := make([]byte, 1)
bufW := bytes.NewBuffer(nil)
for { for {
n, err := conn.Read(bufR) n, err := conn.Read(bufR)
require.NoError(t, err) require.NoError(t, err)
@ -183,16 +239,22 @@ func TCPServer(t *testing.T, wg *sync.WaitGroup, wg2 *sync.WaitGroup, wg3 *sync.
var obj GelfObject var obj GelfObject
err = json.Unmarshal(bufW.Bytes(), &obj) err = json.Unmarshal(bufW.Bytes(), &obj)
require.NoError(t, err) require.NoError(t, err)
assert.Equal(t, obj["short_message"], "telegraf")
assert.Equal(t, obj["_name"], "test1")
assert.Equal(t, obj["_tag1"], "value1")
assert.Equal(t, obj["_value"], float64(1)) assert.Equal(t, obj["_value"], float64(1))
} }
// in TCP scenario only 3 messages are received (1st, 2dn and 4th) due to connection break after the 2nd conn := accept()
defer conn.Close()
recv() // in TCP scenario only 3 messages are received, the 3rd is lost due to simulated connection break after the 2nd
recv()
recv(conn)
recv(conn)
_ = conn.Close() _ = conn.Close()
wg3.Done() wg3.Done()
conn = accept() conn = accept()
defer conn.Close() defer conn.Close()
recv() recv(conn)
} }