mirror of https://github.com/grafana/loki
Send logs to multiple loki instances (#536)
* Adds the ability to provide multiple Loki URL For backward compatibility `client:` still works with flag. * add some tests for multi client * update ksonnet module to support multiple client * fix comment * fix lint issuespull/554/head
Cyril Tovena
6 years ago
committed by
GitHub
parent
fa4f936fcb
commit
53075db577
@ -0,0 +1,57 @@ |
||||
package client |
||||
|
||||
import ( |
||||
"flag" |
||||
"time" |
||||
|
||||
"github.com/cortexproject/cortex/pkg/util" |
||||
"github.com/cortexproject/cortex/pkg/util/flagext" |
||||
"github.com/prometheus/common/model" |
||||
) |
||||
|
||||
// Config describes configuration for a HTTP pusher client.
|
||||
type Config struct { |
||||
URL flagext.URLValue |
||||
BatchWait time.Duration |
||||
BatchSize int |
||||
|
||||
BackoffConfig util.BackoffConfig `yaml:"backoff_config"` |
||||
// The labels to add to any time series or alerts when communicating with loki
|
||||
ExternalLabels model.LabelSet `yaml:"external_labels,omitempty"` |
||||
Timeout time.Duration `yaml:"timeout"` |
||||
} |
||||
|
||||
// RegisterFlags registers flags.
|
||||
func (c *Config) RegisterFlags(flags *flag.FlagSet) { |
||||
flags.Var(&c.URL, "client.url", "URL of log server") |
||||
flags.DurationVar(&c.BatchWait, "client.batch-wait", 1*time.Second, "Maximum wait period before sending batch.") |
||||
flags.IntVar(&c.BatchSize, "client.batch-size-bytes", 100*1024, "Maximum batch size to accrue before sending. ") |
||||
|
||||
flag.IntVar(&c.BackoffConfig.MaxRetries, "client.max-retries", 5, "Maximum number of retires when sending batches.") |
||||
flag.DurationVar(&c.BackoffConfig.MinBackoff, "client.min-backoff", 100*time.Millisecond, "Initial backoff time between retries.") |
||||
flag.DurationVar(&c.BackoffConfig.MaxBackoff, "client.max-backoff", 5*time.Second, "Maximum backoff time between retries.") |
||||
flag.DurationVar(&c.Timeout, "client.timeout", 10*time.Second, "Maximum time to wait for server to respond to a request") |
||||
|
||||
} |
||||
|
||||
// UnmarshalYAML implement Yaml Unmarshaler
|
||||
func (c *Config) UnmarshalYAML(unmarshal func(interface{}) error) error { |
||||
type raw Config |
||||
// force sane defaults.
|
||||
cfg := raw{ |
||||
BackoffConfig: util.BackoffConfig{ |
||||
MaxBackoff: 5 * time.Second, |
||||
MaxRetries: 5, |
||||
MinBackoff: 100 * time.Millisecond, |
||||
}, |
||||
BatchSize: 100 * 1024, |
||||
BatchWait: 1 * time.Second, |
||||
Timeout: 10 * time.Second, |
||||
} |
||||
if err := unmarshal(&cfg); err != nil { |
||||
return err |
||||
} |
||||
|
||||
*c = Config(cfg) |
||||
return nil |
||||
} |
||||
@ -0,0 +1,24 @@ |
||||
package fake |
||||
|
||||
import ( |
||||
"time" |
||||
|
||||
"github.com/grafana/loki/pkg/promtail/api" |
||||
"github.com/prometheus/common/model" |
||||
) |
||||
|
||||
// Client is a fake client used for testing.
|
||||
type Client struct { |
||||
OnHandleEntry api.EntryHandlerFunc |
||||
OnStop func() |
||||
} |
||||
|
||||
// Stop implements client.Client
|
||||
func (c *Client) Stop() { |
||||
c.OnStop() |
||||
} |
||||
|
||||
// Handle implements client.Client
|
||||
func (c *Client) Handle(labels model.LabelSet, time time.Time, entry string) error { |
||||
return c.OnHandleEntry.Handle(labels, time, entry) |
||||
} |
||||
@ -0,0 +1,43 @@ |
||||
package client |
||||
|
||||
import ( |
||||
"errors" |
||||
"time" |
||||
|
||||
"github.com/go-kit/kit/log" |
||||
"github.com/grafana/loki/pkg/util" |
||||
"github.com/prometheus/common/model" |
||||
) |
||||
|
||||
// MultiClient is client pushing to one or more loki instances.
|
||||
type MultiClient []Client |
||||
|
||||
// NewMulti creates a new client
|
||||
func NewMulti(logger log.Logger, cfgs ...Config) (Client, error) { |
||||
if len(cfgs) == 0 { |
||||
return nil, errors.New("at least one client config should be provided") |
||||
} |
||||
var clients []Client |
||||
for _, cfg := range cfgs { |
||||
clients = append(clients, New(cfg, logger)) |
||||
} |
||||
return MultiClient(clients), nil |
||||
} |
||||
|
||||
// Handle Implements api.EntryHandler
|
||||
func (m MultiClient) Handle(labels model.LabelSet, time time.Time, entry string) error { |
||||
var result util.MultiError |
||||
for _, client := range m { |
||||
if err := client.Handle(labels, time, entry); err != nil { |
||||
result.Add(err) |
||||
} |
||||
} |
||||
return result.Err() |
||||
} |
||||
|
||||
// Stop implements Client
|
||||
func (m MultiClient) Stop() { |
||||
for _, c := range m { |
||||
c.Stop() |
||||
} |
||||
} |
||||
@ -0,0 +1,103 @@ |
||||
package client |
||||
|
||||
import ( |
||||
"errors" |
||||
"net/url" |
||||
"reflect" |
||||
"testing" |
||||
"time" |
||||
|
||||
"github.com/grafana/loki/pkg/promtail/api" |
||||
|
||||
"github.com/cortexproject/cortex/pkg/util" |
||||
"github.com/cortexproject/cortex/pkg/util/flagext" |
||||
"github.com/grafana/loki/pkg/promtail/client/fake" |
||||
"github.com/prometheus/common/model" |
||||
) |
||||
|
||||
func TestNewMulti(t *testing.T) { |
||||
_, err := NewMulti(util.Logger, []Config{}...) |
||||
if err == nil { |
||||
t.Fatal("expected err but got nil") |
||||
} |
||||
host1, _ := url.Parse("http://localhost:3100") |
||||
host2, _ := url.Parse("https://grafana.com") |
||||
expectedCfg1 := Config{BatchSize: 20, URL: flagext.URLValue{URL: host1}} |
||||
expectedCfg2 := Config{BatchSize: 10, URL: flagext.URLValue{URL: host2}} |
||||
|
||||
clients, err := NewMulti(util.Logger, expectedCfg1, expectedCfg2) |
||||
if err != nil { |
||||
t.Fatalf("expected err: nil got:%v", err) |
||||
} |
||||
multi := clients.(MultiClient) |
||||
if len(multi) != 2 { |
||||
t.Fatalf("expected client: 2 got:%d", len(multi)) |
||||
} |
||||
cfg1 := clients.(MultiClient)[0].(*client).cfg |
||||
|
||||
if !reflect.DeepEqual(cfg1, expectedCfg1) { |
||||
t.Fatalf("expected cfg: %v got:%v", expectedCfg1, cfg1) |
||||
} |
||||
|
||||
cfg2 := clients.(MultiClient)[1].(*client).cfg |
||||
|
||||
if !reflect.DeepEqual(cfg2, expectedCfg2) { |
||||
t.Fatalf("expected cfg: %v got:%v", expectedCfg2, cfg2) |
||||
} |
||||
} |
||||
|
||||
func TestMultiClient_Stop(t *testing.T) { |
||||
var stopped int |
||||
|
||||
stopping := func() { |
||||
stopped++ |
||||
} |
||||
fc := &fake.Client{OnStop: stopping} |
||||
clients := []Client{fc, fc, fc, fc} |
||||
m := MultiClient(clients) |
||||
|
||||
m.Stop() |
||||
|
||||
if stopped != len(clients) { |
||||
t.Fatal("missing stop call") |
||||
} |
||||
} |
||||
|
||||
func TestMultiClient_Handle(t *testing.T) { |
||||
|
||||
var called int |
||||
|
||||
errorFn := api.EntryHandlerFunc(func(labels model.LabelSet, time time.Time, entry string) error { called++; return errors.New("") }) |
||||
okFn := api.EntryHandlerFunc(func(labels model.LabelSet, time time.Time, entry string) error { called++; return nil }) |
||||
|
||||
errfc := &fake.Client{OnHandleEntry: errorFn} |
||||
okfc := &fake.Client{OnHandleEntry: okFn} |
||||
t.Run("some error", func(t *testing.T) { |
||||
clients := []Client{okfc, errfc, okfc, errfc, errfc, okfc} |
||||
m := MultiClient(clients) |
||||
|
||||
if err := m.Handle(nil, time.Now(), ""); err == nil { |
||||
t.Fatal("expected err got nil") |
||||
} |
||||
|
||||
if called != len(clients) { |
||||
t.Fatal("missing handle call") |
||||
} |
||||
|
||||
}) |
||||
t.Run("no error", func(t *testing.T) { |
||||
called = 0 |
||||
clients := []Client{okfc, okfc, okfc, okfc, okfc, okfc} |
||||
m := MultiClient(clients) |
||||
|
||||
if err := m.Handle(nil, time.Now(), ""); err != nil { |
||||
t.Fatal("expected err to be nil") |
||||
} |
||||
|
||||
if called != len(clients) { |
||||
t.Fatal("missing handle call") |
||||
} |
||||
|
||||
}) |
||||
|
||||
} |
||||
@ -0,0 +1,48 @@ |
||||
package util |
||||
|
||||
import ( |
||||
"bytes" |
||||
"fmt" |
||||
) |
||||
|
||||
// The MultiError type implements the error interface, and contains the
|
||||
// Errors used to construct it.
|
||||
type MultiError []error |
||||
|
||||
// Returns a concatenated string of the contained errors
|
||||
func (es MultiError) Error() string { |
||||
var buf bytes.Buffer |
||||
|
||||
if len(es) > 1 { |
||||
_, _ = fmt.Fprintf(&buf, "%d errors: ", len(es)) |
||||
} |
||||
|
||||
for i, err := range es { |
||||
if i != 0 { |
||||
buf.WriteString("; ") |
||||
} |
||||
buf.WriteString(err.Error()) |
||||
} |
||||
|
||||
return buf.String() |
||||
} |
||||
|
||||
// Add adds the error to the error list if it is not nil.
|
||||
func (es *MultiError) Add(err error) { |
||||
if err == nil { |
||||
return |
||||
} |
||||
if merr, ok := err.(MultiError); ok { |
||||
*es = append(*es, merr...) |
||||
} else { |
||||
*es = append(*es, err) |
||||
} |
||||
} |
||||
|
||||
// Err returns the error list as an error or nil if it is empty.
|
||||
func (es MultiError) Err() error { |
||||
if len(es) == 0 { |
||||
return nil |
||||
} |
||||
return es |
||||
} |
||||
Loading…
Reference in new issue