mirror of https://github.com/grafana/loki
Asynchronous Promtail stages (#2996)
* Introducing go pipelines to promtail. Based off @jeschkies idea. WIP Signed-off-by: Cyril Tovena <cyril.tovena@gmail.com> * Fixes all tests. Signed-off-by: Cyril Tovena <cyril.tovena@gmail.com> * More tests and code cleanup. Signed-off-by: Cyril Tovena <cyril.tovena@gmail.com> * Wip breaking things. Signed-off-by: Cyril Tovena <cyril.tovena@gmail.com> * Fixing tests, adding Stop to the interface. Signed-off-by: Cyril Tovena <cyril.tovena@gmail.com> * Fixes all tests. Signed-off-by: Cyril Tovena <cyril.tovena@gmail.com> * Fixes more test. Signed-off-by: Cyril Tovena <cyril.tovena@gmail.com> * Moar fixes for tests. Signed-off-by: Cyril Tovena <cyril.tovena@gmail.com> * Close correctly client before reading. Signed-off-by: Cyril Tovena <cyril.tovena@gmail.com> * Use defer. Signed-off-by: Cyril Tovena <cyril.tovena@gmail.com> * Add some comments. Signed-off-by: Cyril Tovena <cyril.tovena@gmail.com> * Fixes lint issues Signed-off-by: Cyril Tovena <cyril.tovena@gmail.com> * Attempt to fix journald without linux. Signed-off-by: Cyril Tovena <cyril.tovena@gmail.com> * Fixes journald json test. Signed-off-by: Cyril Tovena <cyril.tovena@gmail.com> * Add missing stop in the filetarget. Signed-off-by: Cyril Tovena <cyril.tovena@gmail.com> * Update pkg/logentry/stages/match_test.go Co-authored-by: Karsten Jeschkies <k@jeschkies.xyz> * First set of feeback review fixes. Signed-off-by: Cyril Tovena <cyril.tovena@gmail.com> * Use newEntry as suggested by Karsten. Signed-off-by: Cyril Tovena <cyril.tovena@gmail.com> * Fix tests. Signed-off-by: Cyril Tovena <cyril.tovena@gmail.com> * lint. Signed-off-by: Cyril Tovena <cyril.tovena@gmail.com> Co-authored-by: Karsten Jeschkies <k@jeschkies.xyz>pull/3025/head
Cyril Tovena
5 years ago
committed by
GitHub
parent
05f3d6e0da
commit
c4faa579e3
@ -1,48 +1,91 @@ |
||||
package api |
||||
|
||||
import ( |
||||
"time" |
||||
"sync" |
||||
|
||||
"github.com/prometheus/common/model" |
||||
|
||||
"github.com/grafana/loki/pkg/logproto" |
||||
) |
||||
|
||||
// Entry is a log entry with labels.
|
||||
type Entry struct { |
||||
Labels model.LabelSet |
||||
logproto.Entry |
||||
} |
||||
|
||||
type InstrumentedEntryHandler interface { |
||||
EntryHandler |
||||
UnregisterLatencyMetric(labels model.LabelSet) |
||||
} |
||||
|
||||
// EntryHandler is something that can "handle" entries.
|
||||
// EntryHandler is something that can "handle" entries via a channel.
|
||||
// Stop must be called to gracefully shutdown the EntryHandler
|
||||
type EntryHandler interface { |
||||
Handle(labels model.LabelSet, time time.Time, entry string) error |
||||
Chan() chan<- Entry |
||||
Stop() |
||||
} |
||||
|
||||
// EntryHandlerFunc is modelled on http.HandlerFunc.
|
||||
type EntryHandlerFunc func(labels model.LabelSet, time time.Time, entry string) error |
||||
|
||||
// Handle implements EntryHandler.
|
||||
func (e EntryHandlerFunc) Handle(labels model.LabelSet, time time.Time, entry string) error { |
||||
return e(labels, time, entry) |
||||
} |
||||
|
||||
// EntryMiddleware is something that takes on EntryHandler and produces another.
|
||||
// EntryMiddleware takes an EntryHandler and returns another one that will intercept and forward entries.
|
||||
// The newly created EntryHandler should be Stopped independently from the original one.
|
||||
type EntryMiddleware interface { |
||||
Wrap(next EntryHandler) EntryHandler |
||||
Wrap(EntryHandler) EntryHandler |
||||
} |
||||
|
||||
// EntryMiddlewareFunc is modelled on http.HandlerFunc.
|
||||
type EntryMiddlewareFunc func(next EntryHandler) EntryHandler |
||||
// EntryMiddlewareFunc allows to create EntryMiddleware via a function.
|
||||
type EntryMiddlewareFunc func(EntryHandler) EntryHandler |
||||
|
||||
// Wrap implements EntryMiddleware.
|
||||
func (e EntryMiddlewareFunc) Wrap(next EntryHandler) EntryHandler { |
||||
return e(next) |
||||
} |
||||
|
||||
// EntryMutatorFunc is a function that can mutate an entry
|
||||
type EntryMutatorFunc func(Entry) Entry |
||||
|
||||
type entryHandler struct { |
||||
stop func() |
||||
entries chan<- Entry |
||||
} |
||||
|
||||
func (e entryHandler) Chan() chan<- Entry { |
||||
return e.entries |
||||
} |
||||
|
||||
func (e entryHandler) Stop() { |
||||
e.stop() |
||||
} |
||||
|
||||
// NewEntryHandler creates a new EntryHandler using a input channel and a stop function.
|
||||
func NewEntryHandler(entries chan<- Entry, stop func()) EntryHandler { |
||||
return entryHandler{ |
||||
stop: stop, |
||||
entries: entries, |
||||
} |
||||
} |
||||
|
||||
// NewEntryMutatorHandler creates a EntryHandler that mutates incoming entries from another EntryHandler.
|
||||
func NewEntryMutatorHandler(next EntryHandler, f EntryMutatorFunc) EntryHandler { |
||||
in, wg, once := make(chan Entry), sync.WaitGroup{}, sync.Once{} |
||||
nextChan := next.Chan() |
||||
wg.Add(1) |
||||
go func() { |
||||
defer wg.Done() |
||||
for e := range in { |
||||
nextChan <- f(e) |
||||
} |
||||
}() |
||||
return NewEntryHandler(in, func() { |
||||
once.Do(func() { close(in) }) |
||||
wg.Wait() |
||||
}) |
||||
} |
||||
|
||||
// AddLabelsMiddleware is an EntryMiddleware that adds some labels.
|
||||
func AddLabelsMiddleware(additionalLabels model.LabelSet) EntryMiddleware { |
||||
return EntryMiddlewareFunc(func(next EntryHandler) EntryHandler { |
||||
return EntryHandlerFunc(func(labels model.LabelSet, time time.Time, entry string) error { |
||||
labels = additionalLabels.Merge(labels) // Add the additionalLabels but preserves the original labels.
|
||||
return next.Handle(labels, time, entry) |
||||
return EntryMiddlewareFunc(func(eh EntryHandler) EntryHandler { |
||||
return NewEntryMutatorHandler(eh, func(e Entry) Entry { |
||||
e.Labels = additionalLabels.Merge(e.Labels) |
||||
return e |
||||
}) |
||||
}) |
||||
} |
||||
|
||||
@ -1,30 +1,58 @@ |
||||
package fake |
||||
|
||||
import ( |
||||
"time" |
||||
|
||||
"github.com/prometheus/common/model" |
||||
"sync" |
||||
|
||||
"github.com/grafana/loki/pkg/promtail/api" |
||||
) |
||||
|
||||
// Client is a fake client used for testing.
|
||||
type Client struct { |
||||
OnHandleEntry api.EntryHandlerFunc |
||||
OnStop func() |
||||
entries chan api.Entry |
||||
received []api.Entry |
||||
once sync.Once |
||||
mtx sync.Mutex |
||||
wg sync.WaitGroup |
||||
OnStop func() |
||||
} |
||||
|
||||
func New(stop func()) *Client { |
||||
c := &Client{ |
||||
OnStop: stop, |
||||
entries: make(chan api.Entry), |
||||
} |
||||
c.wg.Add(1) |
||||
go func() { |
||||
defer c.wg.Done() |
||||
for e := range c.entries { |
||||
c.mtx.Lock() |
||||
c.received = append(c.received, e) |
||||
c.mtx.Unlock() |
||||
} |
||||
}() |
||||
return c |
||||
} |
||||
|
||||
// Stop implements client.Client
|
||||
func (c *Client) Stop() { |
||||
c.once.Do(func() { close(c.entries) }) |
||||
c.wg.Wait() |
||||
c.OnStop() |
||||
} |
||||
|
||||
// StopNow implements client.Client
|
||||
func (c *Client) StopNow() { |
||||
c.OnStop() |
||||
func (c *Client) Chan() chan<- api.Entry { |
||||
return c.entries |
||||
} |
||||
|
||||
func (c *Client) Received() []api.Entry { |
||||
c.mtx.Lock() |
||||
defer c.mtx.Unlock() |
||||
cpy := make([]api.Entry, len(c.received)) |
||||
copy(cpy, c.received) |
||||
return cpy |
||||
} |
||||
|
||||
// Handle implements client.Client
|
||||
func (c *Client) Handle(labels model.LabelSet, time time.Time, entry string) error { |
||||
return c.OnHandleEntry.Handle(labels, time, entry) |
||||
// StopNow implements client.Client
|
||||
func (c *Client) StopNow() { |
||||
c.Stop() |
||||
} |
||||
|
||||
Loading…
Reference in new issue