logging

Documentation

Overview

logging tails a log file in real time, sending new lines to a channel with exponential backoff when waiting for new data.

Log entry types and paginated querying. Reads the log file backwards so the most recent entries come first.

Index

• func MarshalLokiBatch(batch LokiBatch) ([]byte, error)
• func SendLokiTestEntry(endpoint, bearerToken, tenantID string) error
• func TailFile(ctx context.Context, path string, lines chan<- string) error
• type LogEntry
  • func (e *LogEntry) UnmarshalJSON(data []byte) error
• type LogRequest
• type LogTLS
• type LokiBatch
• type LokiBatcher
  • func NewLokiBatcher(lines <-chan TaggedLine, batches chan<- LokiBatch, batchSize int, ...) *LokiBatcher
  • func (b *LokiBatcher) Run(ctx context.Context)
• type LokiEntry
• type LokiPipeline
  • func NewLokiPipeline(store *config.ConfigStore, positionsPath string, resolveSinks SinkResolver) *LokiPipeline
  • func (p *LokiPipeline) GetPusher() *LokiPusher
  • func (p *LokiPipeline) GetStatus() (bool, map[string]SinkStatus)
  • func (p *LokiPipeline) GetTailerNames() []string
  • func (p *LokiPipeline) IsRunning() bool
  • func (p *LokiPipeline) Reconfigure()
  • func (p *LokiPipeline) Restart()
  • func (p *LokiPipeline) Start()
  • func (p *LokiPipeline) Stop()
• type LokiPusher
  • func NewLokiPusher(endpoint, bearerToken, tenantID string, batches <-chan LokiBatch) *LokiPusher
  • func (p *LokiPusher) GetStatus() map[string]SinkStatus
  • func (p *LokiPusher) Run(ctx context.Context)
• type LokiStream
• type LokiTailer
  • func NewLokiTailer(sink, path string, pos *PositionStore, lines chan<- TaggedLine) *LokiTailer
  • func (t *LokiTailer) Run(ctx context.Context)
• type PositionStore
  • func NewPositionStore(path string) *PositionStore
  • func (ps *PositionStore) Cleanup(activePaths map[string]bool)
  • func (ps *PositionStore) Get(path string) int64
  • func (ps *PositionStore) Load() error
  • func (ps *PositionStore) Remove(path string)
  • func (ps *PositionStore) Save() error
  • func (ps *PositionStore) Set(path string, offset int64)
• type QueryParams
• type QueryResult
  • func QueryLogs(path string, params QueryParams) (*QueryResult, error)
• type SinkResolver
• type SinkStatus
• type TaggedLine

Functions

func MarshalLokiBatch

func MarshalLokiBatch(batch LokiBatch) ([]byte, error)

func SendLokiTestEntry

func SendLokiTestEntry(endpoint, bearerToken, tenantID string) error

func TailFile

func TailFile(ctx context.Context, path string, lines chan<- string) error

Types

type LogEntry

type LogEntry struct {
	Level   string      `json:"level"`
	Ts      float64     `json:"ts"`
	Logger  string      `json:"logger"`
	Msg     string      `json:"msg"`
	Request *LogRequest `json:"request,omitempty"`

	Duration    float64             `json:"duration,omitempty"`
	Status      int                 `json:"status,omitempty"`
	Size        int                 `json:"size,omitempty"`
	RespHeaders map[string][]string `json:"resp_headers,omitempty"`

	Extra map[string]any `json:"extra,omitempty"`
}

func (*LogEntry) UnmarshalJSON

func (e *LogEntry) UnmarshalJSON(data []byte) error

type LogRequest

type LogRequest struct {
	RemoteAddr string              `json:"remote_addr"`
	Proto      string              `json:"proto"`
	Method     string              `json:"method"`
	Host       string              `json:"host"`
	URI        string              `json:"uri"`
	Headers    map[string][]string `json:"headers,omitempty"`
	TLS        *LogTLS             `json:"tls,omitempty"`
}

type LogTLS

type LogTLS struct {
	Resumed     bool   `json:"resumed"`
	Version     int    `json:"version"`
	CipherSuite int    `json:"cipher_suite"`
	Proto       string `json:"proto"`
	ServerName  string `json:"server_name"`
}

type LokiBatch

type LokiBatch struct {
	Streams []LokiStream
}

type LokiBatcher

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

func NewLokiBatcher

func NewLokiBatcher(
	lines <-chan TaggedLine,
	batches chan<- LokiBatch,
	batchSize int,
	flushInterval time.Duration,
	staticLabels map[string]string,
) *LokiBatcher

func (*LokiBatcher) Run

func (b *LokiBatcher) Run(ctx context.Context)

type LokiEntry

type LokiEntry struct {
	Timestamp string
	Line      string
}

type LokiPipeline

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

func NewLokiPipeline

func NewLokiPipeline(store *config.ConfigStore, positionsPath string, resolveSinks SinkResolver) *LokiPipeline

func (*LokiPipeline) GetPusher

func (p *LokiPipeline) GetPusher() *LokiPusher

func (*LokiPipeline) GetStatus

func (p *LokiPipeline) GetStatus() (bool, map[string]SinkStatus)

func (*LokiPipeline) GetTailerNames

func (p *LokiPipeline) GetTailerNames() []string

func (*LokiPipeline) IsRunning

func (p *LokiPipeline) IsRunning() bool

func (*LokiPipeline) Reconfigure

func (p *LokiPipeline) Reconfigure()

func (*LokiPipeline) Restart

func (p *LokiPipeline) Restart()

func (*LokiPipeline) Start

func (p *LokiPipeline) Start()

func (*LokiPipeline) Stop

func (p *LokiPipeline) Stop()

type LokiPusher

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

func NewLokiPusher

func NewLokiPusher(
	endpoint, bearerToken, tenantID string,
	batches <-chan LokiBatch,
) *LokiPusher

func (*LokiPusher) GetStatus

func (p *LokiPusher) GetStatus() map[string]SinkStatus

func (*LokiPusher) Run

func (p *LokiPusher) Run(ctx context.Context)

type LokiStream

type LokiStream struct {
	Labels  map[string]string
	Entries []LokiEntry
}

type LokiTailer

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

func NewLokiTailer

func NewLokiTailer(sink, path string, pos *PositionStore, lines chan<- TaggedLine) *LokiTailer

func (*LokiTailer) Run

func (t *LokiTailer) Run(ctx context.Context)

type PositionStore

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

func NewPositionStore

func NewPositionStore(path string) *PositionStore

func (*PositionStore) Cleanup

func (ps *PositionStore) Cleanup(activePaths map[string]bool)

func (*PositionStore) Get

func (ps *PositionStore) Get(path string) int64

func (*PositionStore) Load

func (ps *PositionStore) Load() error

func (*PositionStore) Remove

func (ps *PositionStore) Remove(path string)

func (*PositionStore) Save

func (ps *PositionStore) Save() error

func (*PositionStore) Set

func (ps *PositionStore) Set(path string, offset int64)

type QueryParams

type QueryParams struct {
	Limit     int
	Offset    int
	Level     string
	Host      string
	StatusMin int
	StatusMax int
	Since     time.Time
	Until     time.Time
}

type QueryResult

type QueryResult struct {
	Entries []LogEntry `json:"entries"`
	HasMore bool       `json:"has_more"`
}

func QueryLogs

func QueryLogs(path string, params QueryParams) (*QueryResult, error)

type SinkResolver

type SinkResolver func() map[string]string

type SinkStatus

type SinkStatus struct {
	Tailing       bool      `json:"tailing"`
	LastPushAt    time.Time `json:"last_push_at"`
	EntriesPushed int64     `json:"entries_pushed"`
	LastError     string    `json:"last_error"`
}

type TaggedLine

type TaggedLine struct {
	Sink string
	Line string
}