logtail

README

logtail is a log tailing utility.

1. Features

• tailing command output
• support watch files under directory/sub-directories
• support (dynamically) multiple commands tailing
• support websocket tailing
• support log matching filter
• support transfer log to console/file/webhook (include dingtalk)

2. Architecture

3. Usage

3.1. install logtail

go get -u github.com/vogo/logtail/cmd/logtail@master

3.2. start logtail server

usage: logtail -port=<port> -cmd="<cmd>"

examples:

# tailing file logs
logtail -port=54321 -cmd="tail -F /home/my/logs/myapp.log"

# tailing kubectl logs
logtail -port=54322 -cmd="kubectl logs --tail 10 -f \$(kubectl get pods --selector=app=myapp -o jsonpath='{.items[*].metadata.name}')"

# using a config file
logtail -file=/Users/wongoo/logtail-config.json

config file sample:

{
  "port": 54321,
  "default_format":{
    "prefix": "!!!!-!!-!!"
  },
  "global_routers": [
    {
      "matchers": [],
      "transfers": [
        {"type": "console"}
      ]
    }
  ],
  "default_routers": [
    {
      "matchers": [],
      "transfers": [
        {"type": "console"}
      ]
    }
  ],
  "servers": [
    {
      "id": "app1",
      "command": "tail -f /Users/wongoo/app/app1.log",
      "routers": [
        {
          "matchers": [
            {"match_contains": "ERROR"}
          ],
          "transfers": [
            {
              "type": "ding",
              "ding_url": "https://oapi.dingtalk.com/robot/send?access_token=<token>"
            },
            {
              "type": "webhook",
              "webhook_url": "http://127.0.0.1:9000"
            },
            {
              "type": "file",
              "dir": "/opt/logs/"
            }
          ]
        }
      ]
    },
    {
      "id": "app2",
      "command": "tail -f /Users/wongoo/app/app2.log"
    }
  ]
}

Tailing multiple commands (split by new line char \n):

{
  "servers": [
    {
      "id": "app1",
      "commands": "tail -f /Users/wongoo/app/app1.log\ntail -f /Users/wongoo/app/app2.log\ntail -f /Users/wongoo/app/app3.log"
    }
  ]
}

Tailing multiple commands which are generated dynamically:

{
  "servers": [
    {
      "id": "app1",
      "command_gen": "cmd='';for d in $(ls /logs/k8s_logs/service/*.log); do cmd=$cmd'tail -f '$d$'\n'; done;cmd=${cmd::-1}; echo \"$cmd\"",
    }
  ]
}

Watch changing files under directory/sub-directories and tailing them:

{
  "servers": [
    {
      "id": "app1",
      "file": { "path": "/logs/k8s_logs/service-app1/", "recursive": true, "suffix": ".log","method":"timer"}
    },
    {
      "id": "app2",
      "file": { "path": "/logs/k8s_logs/service-app2/", "recursive": true, "suffix": ".log","method":"os"}
    }
  ]
}

The value of method could be os or timer:

• os: using os file system api to monitor file changes
• timer: interval check file stat to check file changes

3.3. tailing logs

browse http://<server-ip>:<port> to list all tailing logs.

4. log format

You can config log format globally, or config it for a server.

The config prefix of the format is the wildcard of the prefix of a new log record, logtail will check whether a new line is the start of a new log record, or one of the following lines.

The wildcard does NOT support '*' for none or many chars, it supports the following tag:

• '?' as one byte char;
• '~' as one alphabet char;
• '!' as one number char;
• other chars must exact match.

example:

{
    "default_format":{
      "prefix": "!!!!-!!-!!"  # global format config, matches 2020-12-12
    },
    "servers": [
        {
          "id": "app1",
          "command": "tail -f /Users/wongoo/app/app1.log",
          "format":{
            "prefix": "!!!!-!!-!!" # server format config, matches 2020-12-12
          }
        }
    ]
}

5. command examples

# tail log file
tail -f /usr/local/myapp/myapp.log

# k8s: find and tail logs for the single pod of myapp
kubectl logs --tail 10 -f $(kubectl get pods --selector=app=myapp -o jsonpath='{.items[*].metadata.name}')

# k8s: find and tail logs for the myapp deployment (multiple pods)
kubectl logs --tail 10 -f deployment/$(kubectl get deployments --selector=project-name=myapp -o jsonpath='{.items[*].metadata.name}')

# k8s: find and tail logs for the latest version of the myapp deployment (single pod)
s=$(kubectl get deployments --selector=project-name=myapp -o jsonpath='{.items[*].metadata.name}');s=${s##* };kubectl logs --tail 10 -f deployment/$s

# k8s: find and tail logs for the latest version of the myapp deployment (multiple pods)
app=$(kubectl get deployments --selector=project-name=myapp -o jsonpath='{.items[*].metadata.name}');app=${app##* };pods=$(kubectl get pods --selector=app=$app -o jsonpath='{.items[*].metadata.name}');cmd='';for pod in $pods; do cmd=$cmd'kubectl logs --tail 2 -f pod/'$pod$'\n'; done;cmd=${cmd::-1}; echo "$cmd"

Documentation

Index

• Constants
• Variables
• func EscapeLimitJSONBytes(b []byte, capacity int) []byte
• func Start()
• func StartLogtail(config *Config)
• func StopLogtail()
• func WildcardMatch(pattern string, data []byte) bool
• type Channel
• type Config
• type ConsoleTransfer
  • func (d *ConsoleTransfer) Trans(serverID string, data ...[]byte) error
• type ContainsMatcher
  • func NewContainsMatcher(pattern string, contains bool) *ContainsMatcher
  • func (cm *ContainsMatcher) Match(bytes []byte) bool
• type DingTransfer
  • func (d *DingTransfer) Trans(serverID string, data ...[]byte) error
• type FileConfig
• type FileTransfer
  • func (ft *FileTransfer) Trans(serverID string, data ...[]byte) error
• type Filter
  • func (f *Filter) Match(bytes []byte, length, index *int) []byte
  • func (f *Filter) Route(bytes []byte) error
  • func (f *Filter) Trans(bytes ...[]byte) error
• type Format
  • func (f *Format) PrefixMatch(data []byte) bool
  • func (f *Format) String() string
• type Matcher
• type MatcherConfig
• type Router
  • func NewRouter(s *Server, matchers []Matcher, transfers []Transfer) *Router
  • func (r *Router) SetMatchers(matchers []Matcher)
  • func (r *Router) Start() error
  • func (r *Router) Stop()
• type RouterConfig
• type Server
  • func NewServer(config *Config, serverConfig *ServerConfig) *Server
  • func (s *Server) Fire(data []byte) error
  • func (s *Server) Start()
  • func (s *Server) Stop() error
  • func (s *Server) Write(data []byte) (int, error)
• type ServerConfig
• type Transfer
  • func NewDingTransfer(url string) Transfer
  • func NewFileTransfer(dir string) Transfer
  • func NewWebhookTransfer(url string) Transfer
• type TransferConfig
• type WebhookTransfer
  • func (d *WebhookTransfer) Trans(_ string, data ...[]byte) error
• type WebsocketTransfer
  • func (ww *WebsocketTransfer) Trans(_ string, data ...[]byte) (err error)

Constants

const (
	// URIIndexPrefix uri index prefix.
	URIIndexPrefix = "/index"

	// URITailPrefix uri tail prefix.
	URITailPrefix = "/tail"
)

const (
	// DefaultServerID default server id.
	DefaultServerID = "default"

	// CommandFailRetryInterval command fail retry interval.
	CommandFailRetryInterval = 10 * time.Second
)

const DefaultChannelBufferSize = 16

const DefaultServerPort = 54321

const DoubleSize = 2

const DurationReadNextTimeout = time.Millisecond * 60

const MessageTypeMatcherConfig = '1'

const TransferFileSize = 1024 * 1024 * 8

8MB for each transfer file.

const TransferTypeConsole = "console"

const TransferTypeDing = "ding"

const TransferTypeFile = "file"

const TransferTypeWebhook = "webhook"

const WebsocketHeartbeatReadTimeout = 15 * time.Second

Variables

var (
	ErrNoServerConfig   = errors.New("no server config")
	ErrServerIDNil      = errors.New("server id is nil")
	ErrNoTailingConfig  = errors.New("no tailing command/file config")
	ErrTransURLNil      = errors.New("transfer url is nil")
	ErrTransTypeNil     = errors.New("transfer type is nil")
	ErrTransTypeInvalid = errors.New("invalid transfer type")
	ErrTransDirNil      = errors.New("transfer dir is nil")
)

var ErrHTTPStatusNonOK = errors.New("http status non ok")

var ErrWorkerCommandStopped = errors.New("worker command stopped")

Functions

func EscapeLimitJSONBytes

func EscapeLimitJSONBytes(b []byte, capacity int) []byte

func Start

func Start()

Start parse command config, and start logtail servers with http listener.

func StartLogtail

func StartLogtail(config *Config)

StartLogtail start config servers.

func StopLogtail

func StopLogtail()

StopLogtail stop servers.

func WildcardMatch

func WildcardMatch(pattern string, data []byte) bool

WildcardMatch - finds whether the bytes matches/satisfies the pattern wildcard. supports: - '?' as one byte char - '~' as one alphabet char - '!' as one number char NOT support '*' for none or many char.

Types

type Channel

type Channel chan []byte

type Config

type Config struct {
	Port           int             `json:"port"`
	LogLevel       string          `json:"log_level"`
	DefaultFormat  *Format         `json:"default_format"`
	Servers        []*ServerConfig `json:"servers"`
	DefaultRouters []*RouterConfig `json:"default_routers"`
	GlobalRouters  []*RouterConfig `json:"global_routers"`
}

type ConsoleTransfer

type ConsoleTransfer struct{}

func (*ConsoleTransfer) Trans

func (d *ConsoleTransfer) Trans(serverID string, data ...[]byte) error

type ContainsMatcher

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

func NewContainsMatcher

func NewContainsMatcher(pattern string, contains bool) *ContainsMatcher

func (*ContainsMatcher) Match

func (cm *ContainsMatcher) Match(bytes []byte) bool

type DingTransfer

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

func (*DingTransfer) Trans

func (d *DingTransfer) Trans(serverID string, data ...[]byte) error

type FileConfig

type FileConfig struct {
	// Path the file or directory to tail.
	Path string `json:"path"`

	// Method watch method,
	// - os: using os file system api to monitor file changes,
	// - timer: interval check file stat to check file changes,
	// For some network mount devices, can't get file change events for os api,
	// you'd better to check file stat to know the changes.
	Method fwatch.WatchMethod `json:"method"`

	// only tailing files with the prefix.
	Prefix string `json:"prefix"`

	// only tailing files with the suffix.
	Suffix string `json:"suffix"`

	// Whether include all files in sub directories recursively.
	Recursive bool `json:"recursive"`
}

FileConfig tailing file config.

type FileTransfer

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

func (*FileTransfer) Trans

func (ft *FileTransfer) Trans(serverID string, data ...[]byte) error

type Filter

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

func (*Filter) Match

func (f *Filter) Match(bytes []byte, length, index *int) []byte

func (*Filter) Route

func (f *Filter) Route(bytes []byte) error

func (*Filter) Trans

func (f *Filter) Trans(bytes ...[]byte) error

type Format

type Format struct {
	Prefix string `json:"prefix"` // the wildcard of the line prefix of a log record
}

Format the log format.

func (*Format) PrefixMatch

func (f *Format) PrefixMatch(data []byte) bool

PrefixMatch whether the given data has a prefix of a new record.

func (*Format) String

func (f *Format) String() string

String format string info.

type Matcher

type Matcher interface {
	Match(bytes []byte) bool
}

type MatcherConfig

type MatcherConfig struct {
	Contains    []string `json:"contains"`
	NotContains []string `json:"not_contains"`
}

type Router

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

func NewRouter

func NewRouter(s *Server, matchers []Matcher, transfers []Transfer) *Router

func (*Router) SetMatchers

func (r *Router) SetMatchers(matchers []Matcher)

func (*Router) Start

func (r *Router) Start() error

func (*Router) Stop

func (r *Router) Stop()

type RouterConfig

type RouterConfig struct {
	Matchers  []*MatcherConfig  `json:"matchers"`
	Transfers []*TransferConfig `json:"transfers"`
}

type Server

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

func NewServer

func NewServer(config *Config, serverConfig *ServerConfig) *Server

NewServer start a new server.

func (*Server) Fire

func (s *Server) Fire(data []byte) error

Fire custom generate bytes data to the first worker of the server.

func (*Server) Start

func (s *Server) Start()

Start start server. First, start all routers. Then, call the start func.

func (*Server) Stop

func (s *Server) Stop() error

Stop stop server.

func (*Server) Write

func (s *Server) Write(data []byte) (int, error)

Write bytes data to default workers, which will be send to web socket clients.

type ServerConfig

type ServerConfig struct {
	ID      string          `json:"id"`
	Format  *Format         `json:"format"`
	Routers []*RouterConfig `json:"routers"`

	// single command.
	Command string `json:"command"`

	// multiple commands split by new line.
	Commands string `json:"commands"`

	// command to generate multiple commands split by new line.
	CommandGen string `json:"command_gen"`

	// command to generate multiple commands split by new line.
	File *FileConfig `json:"file"`
}

type Transfer

type Transfer interface {
	Trans(serverID string, data ...[]byte) error
	// contains filtered or unexported methods
}

func NewDingTransfer

func NewDingTransfer(url string) Transfer

func NewFileTransfer

func NewFileTransfer(dir string) Transfer

func NewWebhookTransfer

func NewWebhookTransfer(url string) Transfer

type TransferConfig

type TransferConfig struct {
	Type string `json:"type"`
	URL  string `json:"url"`
	Dir  string `json:"dir"`
}

type WebhookTransfer

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

func (*WebhookTransfer) Trans

func (d *WebhookTransfer) Trans(_ string, data ...[]byte) error

type WebsocketTransfer

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

func (*WebsocketTransfer) Trans

func (ww *WebsocketTransfer) Trans(_ string, data ...[]byte) (err error)