sources

Documentation

Index

• type LogSource
  • func NewLogSource(name string, cfg *config.LogsConfig) *LogSource
  • func (s *LogSource) AddInput(input string)
  • func (s *LogSource) Dump(multiline bool) string
  • func (s *LogSource) GetInfo(key string) status.InfoProvider
  • func (s *LogSource) GetInfoStatus() map[string][]string
  • func (s *LogSource) GetInputs() []string
  • func (s *LogSource) GetSourceType() SourceType
  • func (s *LogSource) HideFromStatus()
  • func (s *LogSource) IsHiddenFromStatus() bool
  • func (s *LogSource) RecordBytes(n int64)
  • func (s *LogSource) RegisterInfo(i status.InfoProvider)
  • func (s *LogSource) RemoveInput(input string)
  • func (s *LogSource) SetSourceType(sourceType SourceType)
• type LogSources
  • func NewLogSources() *LogSources
  • func (s *LogSources) AddSource(source *LogSource)
  • func (s *LogSources) GetAddedForType(sourceType string) chan *LogSource
  • func (s *LogSources) GetSources() []*LogSource
  • func (s *LogSources) RemoveSource(source *LogSource)
  • func (s *LogSources) SubscribeAll() (added chan *LogSource, removed chan *LogSource)
  • func (s *LogSources) SubscribeForType(sourceType string) (added chan *LogSource, removed chan *LogSource)
• type ReplaceableSource
  • func NewReplaceableSource(source *LogSource) *ReplaceableSource
  • func (r *ReplaceableSource) AddInput(input string)
  • func (r *ReplaceableSource) Config() *config.LogsConfig
  • func (r *ReplaceableSource) GetInfo(key string) status.InfoProvider
  • func (r *ReplaceableSource) GetSourceType() SourceType
  • func (r *ReplaceableSource) RecordBytes(n int64)
  • func (r *ReplaceableSource) RegisterInfo(i status.InfoProvider)
  • func (r *ReplaceableSource) RemoveInput(input string)
  • func (r *ReplaceableSource) Replace(source *LogSource)
  • func (r *ReplaceableSource) Status() *status.LogStatus
  • func (r *ReplaceableSource) UnderlyingSource() *LogSource
• type SourceType

Types

type LogSource

type LogSource struct {
	Name   string
	Config *config.LogsConfig
	Status *status.LogStatus

	Messages *config.Messages

	// In the case that the source is overridden, keep a reference to the parent for bubbling up information about the child
	ParentSource *LogSource
	// LatencyStats tracks internal stats on the time spent by messages from this source in a processing pipeline, i.e.
	// the duration between when a message is decoded by the tailer/listener/decoder and when the message is handled by a sender
	LatencyStats *statstracker.Tracker
	BytesRead    *status.CountInfo
	// contains filtered or unexported fields
}

LogSource holds a reference to an integration name and a log configuration, and allows to track errors and successful operations on it. Both name and configuration are static for now and determined at creation time. Changing the status is designed to be thread safe.

func NewLogSource

func NewLogSource(name string, cfg *config.LogsConfig) *LogSource

NewLogSource creates a new log source.

func (*LogSource) AddInput

func (s *LogSource) AddInput(input string)

AddInput registers an input as being handled by this source.

func (*LogSource) Dump

func (s *LogSource) Dump(multiline bool) string

Dump provides a dump of the LogSource contents, for debugging purposes. If multiline is true, the result contains newlines for readability.

func (*LogSource) GetInfo

func (s *LogSource) GetInfo(key string) status.InfoProvider

GetInfo gets an InfoProvider instance by the key

func (*LogSource) GetInfoStatus

func (s *LogSource) GetInfoStatus() map[string][]string

GetInfoStatus returns a primitive representation of the info for the status page

func (*LogSource) GetInputs

func (s *LogSource) GetInputs() []string

GetInputs returns the inputs handled by this source.

func (*LogSource) GetSourceType

func (s *LogSource) GetSourceType() SourceType

GetSourceType returns the sourceType used by this source

func (*LogSource) HideFromStatus

func (s *LogSource) HideFromStatus()

HideFromStatus hides the source from the status output

func (*LogSource) IsHiddenFromStatus

func (s *LogSource) IsHiddenFromStatus() bool

IsHiddenFromStatus returns true if this source should be hidden from the status output

func (*LogSource) RecordBytes

func (s *LogSource) RecordBytes(n int64)

RecordBytes reports bytes to the source expvars Since `container_collect_all` reports all docker logs as a single source (even though the source is overridden internally), we need to report the byte count to the parent source used to populate the status page.

func (*LogSource) RegisterInfo

func (s *LogSource) RegisterInfo(i status.InfoProvider)

RegisterInfo registers some info to display on the status page

func (*LogSource) RemoveInput

func (s *LogSource) RemoveInput(input string)

RemoveInput removes an input from this source.

func (*LogSource) SetSourceType

func (s *LogSource) SetSourceType(sourceType SourceType)

SetSourceType sets a format that give information on how the source lines should be parsed

type LogSources

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

LogSources serves as the interface between Schedulers and Launchers, distributing notifications of added/removed LogSources to subscribed Launchers.

Each subscription receives its own unbuffered channel for sources, and should consume from the channel quickly to avoid blocking other goroutines. There is no means to unsubscribe.

If any sources have been added when GetAddedForType is called, then those sources are immediately sent to the channel.

This type is threadsafe, and all of its methods can be called concurrently.

func NewLogSources

func NewLogSources() *LogSources

NewLogSources creates a new log sources.

func (*LogSources) AddSource

func (s *LogSources) AddSource(source *LogSource)

AddSource adds a new source.

All of the subscribers registered for this source's type (src.Config.Type) will be notified.

func (*LogSources) GetAddedForType

func (s *LogSources) GetAddedForType(sourceType string) chan *LogSource

GetAddedForType returns a channel carrying notifications of new sources with the given type.

Any sources added before this call are delivered from a new goroutine.

func (*LogSources) GetSources

func (s *LogSources) GetSources() []*LogSource

GetSources returns all the sources currently held. The result is copied and will not be modified after it is returned. However, the copy in the LogSources instance may change in that time (changing indexes or adding/removing entries).

func (*LogSources) RemoveSource

func (s *LogSources) RemoveSource(source *LogSource)

RemoveSource removes a source.

All of the subscribers registered for this source's type (src.Config.Type) will be notified of its removal.

func (*LogSources) SubscribeAll

func (s *LogSources) SubscribeAll() (added chan *LogSource, removed chan *LogSource)

SubscribeAll returns two channels carrying notifications of all added and removed sources, respectively. This guarantees consistency if sources are added or removed concurrently.

Any sources added before this call are delivered from a new goroutine.

func (*LogSources) SubscribeForType

func (s *LogSources) SubscribeForType(sourceType string) (added chan *LogSource, removed chan *LogSource)

SubscribeForType returns two channels carrying notifications of added and removed sources with the given type, respectively. This guarantees consistency if sources are added or removed concurrently.

Any sources added before this call are delivered from a new goroutine.

type ReplaceableSource

type ReplaceableSource struct {
	sync.RWMutex
	// contains filtered or unexported fields
}

ReplaceableSource is a thread safe wrapper for a LogSource that allows it to be replaced with a new one. There are some uncommon circumstances where a source needs to be replaced on an active tailer. This wrapper helps ensure there is not any unsafe access to the many underlying properties in a LogSource.

func NewReplaceableSource

func NewReplaceableSource(source *LogSource) *ReplaceableSource

NewReplaceableSource returns a new ReplaceableSource

func (*ReplaceableSource) AddInput

func (r *ReplaceableSource) AddInput(input string)

AddInput registers an input as being handled by this source.

func (*ReplaceableSource) Config

func (r *ReplaceableSource) Config() *config.LogsConfig

Config gets the underlying config

func (*ReplaceableSource) GetInfo

func (r *ReplaceableSource) GetInfo(key string) status.InfoProvider

GetInfo gets an InfoProvider instance by the key

func (*ReplaceableSource) GetSourceType

func (r *ReplaceableSource) GetSourceType() SourceType

GetSourceType gets the source type

func (*ReplaceableSource) RecordBytes

func (r *ReplaceableSource) RecordBytes(n int64)

RecordBytes reports bytes to the source expvars

func (*ReplaceableSource) RegisterInfo

func (r *ReplaceableSource) RegisterInfo(i status.InfoProvider)

RegisterInfo registers some info to display on the status page

func (*ReplaceableSource) RemoveInput

func (r *ReplaceableSource) RemoveInput(input string)

RemoveInput removes an input from this source.

func (*ReplaceableSource) Replace

func (r *ReplaceableSource) Replace(source *LogSource)

Replace replaces the source with a new one

func (*ReplaceableSource) Status

func (r *ReplaceableSource) Status() *status.LogStatus

Status gets the underlying status

func (*ReplaceableSource) UnderlyingSource

func (r *ReplaceableSource) UnderlyingSource() *LogSource

UnderlyingSource gets the underlying log source

type SourceType

type SourceType string

SourceType used for log line parsing logic. TODO: remove this logic.

const (
	// DockerSourceType docker source type
	DockerSourceType SourceType = "docker"
	// KubernetesSourceType kubernetes source type
	KubernetesSourceType SourceType = "kubernetes"
	// IntegrationSourceType integration source type
	IntegrationSourceType SourceType = "integration"
)