lib

Documentation

Index

• func AppendError(err error, other error) error
• func DeregisterDestination(name string)
• func DeregisterSource(name string)
• func DestinationsAvailable() (destinations []string)
• func RegisterDestination(name string, destination Destination)
• func RegisterSource(name string, source Source)
• func SourcesAvailable() (sources []string)
• type Destination
  • func GetDestination(name string) (destination Destination)
  • func GetDestinations(names ...string) (destinations []Destination)
• type DestinationFunc
  • func (f DestinationFunc) Close(group string, stream string)
  • func (f DestinationFunc) Open(group string, stream string) (Writer, error)
• type ErrorList
  • func (err ErrorList) Error() string
• type Group
  • func NewGroup(name string, now time.Time) *Group
  • func (group *Group) Add(msg Message, now time.Time) (stream *Stream)
  • func (group *Group) ForEach(f func(*Stream))
  • func (group *Group) HasExpired(timeout time.Duration, now time.Time) bool
  • func (group *Group) Name() string
  • func (group *Group) RemoveExpired(timeout time.Duration, now time.Time) (streams []*Stream)
  • func (group *Group) String() string
• type LogHandler
  • func (h *LogHandler) HandleLog(entry *log.Entry) (err error)
• type LogLevel
  • func (lvl LogLevel) Get() interface{}
  • func (lvl *LogLevel) Set(s string) error
  • func (lvl LogLevel) String() string
• type Message
  • func (m Message) Bytes() []byte
  • func (m Message) ContentLength() int
  • func (m Message) String() string
• type MessageBatch
  • func (list MessageBatch) Len() int
  • func (list MessageBatch) Less(i int, j int) bool
  • func (list MessageBatch) Swap(i int, j int)
• type MessageQueue
  • func NewMessageQueue() *MessageQueue
  • func (q *MessageQueue) Flush() (batch MessageBatch)
  • func (q *MessageQueue) Notify()
  • func (q *MessageQueue) Push(msg Message)
• type Reader
  • func NewMessageDecoder(r io.Reader) Reader
• type Source
  • func GetSource(name string) (source Source)
  • func GetSources(names ...string) (sources []Source)
• type SourceFunc
  • func (f SourceFunc) Open() (Reader, error)
• type Store
  • func NewStore() *Store
  • func (store *Store) Add(msg Message, now time.Time) (group *Group, stream *Stream)
  • func (store *Store) ForEach(f func(*Group))
  • func (store *Store) RemoveExpired(timeout time.Duration, now time.Time) (streams []*Stream)
• type Stream
  • func NewStream(group string, name string, now time.Time) *Stream
  • func (stream *Stream) Add(msg Message, now time.Time)
  • func (stream *Stream) Flush(limits StreamLimits, now time.Time) (list MessageBatch, reason string)
  • func (stream *Stream) Group() string
  • func (stream *Stream) HasExpired(timeout time.Duration, now time.Time) bool
  • func (stream *Stream) Name() string
  • func (stream *Stream) String() string
• type StreamLimits
• type Writer
  • func NewMessageEncoder(w io.Writer) Writer

Functions

func AppendError

func AppendError(err error, other error) error

func DeregisterDestination

func DeregisterDestination(name string)

func DeregisterSource

func DeregisterSource(name string)

func DestinationsAvailable

func DestinationsAvailable() (destinations []string)

func RegisterDestination

func RegisterDestination(name string, destination Destination)

func RegisterSource

func RegisterSource(name string, source Source)

func SourcesAvailable

func SourcesAvailable() (sources []string)

Types

type Destination

type Destination interface {
	Open(group string, stream string) (Writer, error)

	Close(group string, stream string)
}

func GetDestination

func GetDestination(name string) (destination Destination)

func GetDestinations

func GetDestinations(names ...string) (destinations []Destination)

type DestinationFunc

type DestinationFunc func(group string, stream string) (Writer, error)

func (DestinationFunc) Close

func (f DestinationFunc) Close(group string, stream string)

func (DestinationFunc) Open

func (f DestinationFunc) Open(group string, stream string) (Writer, error)

type ErrorList

type ErrorList []error

func (ErrorList) Error

func (err ErrorList) Error() string

type Group

type Group struct {
	// contains filtered or unexported fields
}

func NewGroup

func NewGroup(name string, now time.Time) *Group

func (*Group) Add

func (group *Group) Add(msg Message, now time.Time) (stream *Stream)

func (*Group) ForEach

func (group *Group) ForEach(f func(*Stream))

func (*Group) HasExpired

func (group *Group) HasExpired(timeout time.Duration, now time.Time) bool

func (*Group) Name

func (group *Group) Name() string

func (*Group) RemoveExpired

func (group *Group) RemoveExpired(timeout time.Duration, now time.Time) (streams []*Stream)

func (*Group) String

func (group *Group) String() string

type LogHandler

type LogHandler struct {
	Group    string
	Stream   string
	Hostname string
	Queue    *MessageQueue
}

func (*LogHandler) HandleLog

func (h *LogHandler) HandleLog(entry *log.Entry) (err error)

type LogLevel

type LogLevel log.Level

func (LogLevel) Get

func (lvl LogLevel) Get() interface{}

func (*LogLevel) Set

func (lvl *LogLevel) Set(s string) error

func (LogLevel) String

func (lvl LogLevel) String() string

type Message

type Message struct {
	Group  string        `json:"group,omitempty"`
	Stream string        `json:"stream,omitempty"`
	Event  ecslogs.Event `json:"event,omitempty"`
}

func (Message) Bytes

func (m Message) Bytes() []byte

func (Message) ContentLength

func (m Message) ContentLength() int

func (Message) String

func (m Message) String() string

type MessageBatch

type MessageBatch []Message

func (MessageBatch) Len

func (list MessageBatch) Len() int

func (MessageBatch) Less

func (list MessageBatch) Less(i int, j int) bool

func (MessageBatch) Swap

func (list MessageBatch) Swap(i int, j int)

type MessageQueue

type MessageQueue struct {
	C <-chan struct{}
	// contains filtered or unexported fields
}

func NewMessageQueue

func NewMessageQueue() *MessageQueue

func (*MessageQueue) Flush

func (q *MessageQueue) Flush() (batch MessageBatch)

func (*MessageQueue) Notify

func (q *MessageQueue) Notify()

func (*MessageQueue) Push

func (q *MessageQueue) Push(msg Message)

type Reader

type Reader interface {
	io.Closer

	ReadMessage() (Message, error)
}

func NewMessageDecoder

func NewMessageDecoder(r io.Reader) Reader

type Source

type Source interface {
	Open() (Reader, error)
}

func GetSource

func GetSource(name string) (source Source)

func GetSources

func GetSources(names ...string) (sources []Source)

type SourceFunc

type SourceFunc func() (Reader, error)

func (SourceFunc) Open

func (f SourceFunc) Open() (Reader, error)

type Store

type Store struct {
	// contains filtered or unexported fields
}

func NewStore

func NewStore() *Store

func (*Store) Add

func (store *Store) Add(msg Message, now time.Time) (group *Group, stream *Stream)

func (*Store) ForEach

func (store *Store) ForEach(f func(*Group))

func (*Store) RemoveExpired

func (store *Store) RemoveExpired(timeout time.Duration, now time.Time) (streams []*Stream)

type Stream

type Stream struct {
	// contains filtered or unexported fields
}

func NewStream

func NewStream(group string, name string, now time.Time) *Stream

func (*Stream) Add

func (stream *Stream) Add(msg Message, now time.Time)

func (*Stream) Flush

func (stream *Stream) Flush(limits StreamLimits, now time.Time) (list MessageBatch, reason string)

func (*Stream) Group

func (stream *Stream) Group() string

func (*Stream) HasExpired

func (stream *Stream) HasExpired(timeout time.Duration, now time.Time) bool

func (*Stream) Name

func (stream *Stream) Name() string

func (*Stream) String

func (stream *Stream) String() string

type StreamLimits

type StreamLimits struct {
	MaxCount int
	MaxBytes int
	MaxTime  time.Duration
	Force    bool
}

type Writer

type Writer interface {
	io.Closer

	WriteMessage(Message) error

	WriteMessageBatch(MessageBatch) error
}

func NewMessageEncoder

func NewMessageEncoder(w io.Writer) Writer