webtail

README

webtail

Tail [log]files via websocket

English | Pусский | 日本語

---

webtail is a web-service and golang package used for continious updated files publication via websocker to browser.

Install

go get -v github.com/LeKovr/webtail/...

Download binary

See Latest release

Docker

Starting from 0.43.2 docker images are published at GitHub Packages, so use

docker pull ghcr.io/lekovr/webtail:latest

See docker-compose.yml for usage example.

v0.43.1 is the last version available at dockerhub.

Use package in your service

package main
import (
    "github.com/LeKovr/webtail"
)

func main() {
    wt, err := webtail.New(log, cfg)
    if err != nil {
        return
    }
    go wt.Run()
    defer wt.Close()
    // ...
    http.Handle("/tail", wt)
}

See also: app.go

Note about gorilla/websocket

Starting from v0.30 this code is based on gorilla/websocket chat example. See {client,hub}.go

License

The MIT License (MIT), see LICENSE.

Copyright (c) 2016-2021 Aleksey Kovrizhkin lekovr+webtail@gmail.com

Documentation

Overview

Package webtail holds tailer service You don't need anything except Service methods

Index

• Constants
• func FileServer(path string) http.Handler
• type Client
• type Config
• type Hub
  • func NewHub(logger logr.Logger, ts *TailService, wg *sync.WaitGroup) *Hub
  • func (h *Hub) Close()
  • func (h *Hub) Run()
• type InMessage
• type IndexItemAttr
• type IndexItemAttrStore
• type IndexItemEvent
• type IndexMessage
• type Message
• type Service
  • func New(log logr.Logger, cfg *Config) (*Service, error)
  • func (wt *Service) Close()
  • func (wt *Service) Run()
  • func (wt *Service) ServeHTTP(w http.ResponseWriter, r *http.Request)
• type StatsMessage
• type TailAttr
• type TailMessage
• type TailService
  • func NewTailService(logger logr.Logger, cfg *Config) (*TailService, error)
  • func (ts *TailService) ChannelExists(channel string) bool
  • func (ts *TailService) IndexItem(key string) *IndexItemAttr
  • func (ts *TailService) IndexKeys() []string
  • func (ts *TailService) IndexUpdate(msg *IndexItemEvent)
  • func (ts *TailService) IndexerRun(out chan *IndexItemEvent, wg *sync.WaitGroup)
  • func (ts *TailService) SetTrace(mode string)
  • func (ts *TailService) TailerAppend(channel string, data []byte) bool
  • func (ts *TailService) TailerBuffer(channel string) [][]byte
  • func (ts *TailService) TailerRun(channel string, out chan *TailMessage, readyChan chan struct{}, ...) error
  • func (ts *TailService) TraceEnabled() bool
  • func (ts *TailService) WorkerExists(channel string) bool
  • func (ts *TailService) WorkerStop(channel string)
• type TraceMessage

Constants

const (
	MsgSubscribed        = "success"
	MsgUnSubscribed      = "success"
	MsgUnknownChannel    = "unknown channel"
	MsgNotSubscribed     = "not subscribed"
	MsgWorkerError       = "worker create error"
	MsgSubscribedAlready = "attached already"
	MsgNone              = ""
)

Returned Messages

Functions

func FileServer

func FileServer(path string) http.Handler

FileServer return embedded or given fs

Types

type Client

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

Client is a middleman between the websocket connection and the hub.

type Config

type Config struct {
	Root        string `long:"root"  default:"log/"  description:"Root directory for log files"`
	Bytes       int64  `long:"bytes" default:"5000"  description:"tail from the last Nth location"`
	Lines       int    `long:"lines" default:"100"   description:"keep N old lines for new consumers"`
	MaxLineSize int    `long:"split" default:"180"   description:"split line if longer"`
	ListCache   int    `long:"cache" default:"2"      description:"Time to cache file listing (sec)"`
	Poll        bool   `long:"poll"  description:"use polling, instead of inotify"`
	Trace       bool   `long:"trace" description:"trace worker channels"`

	ClientBufferSize  int `long:"out_buf"      default:"256"  description:"Client Buffer Size"`
	WSReadBufferSize  int `long:"ws_read_buf"  default:"1024" description:"WS Read Buffer Size"`
	WSWriteBufferSize int `long:"ws_write_buf" default:"1024" description:"WS Write Buffer Size"`
}

Config defines local application flags

type Hub

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

Hub maintains the set of active clients and broadcasts messages to them

func NewHub

func NewHub(logger logr.Logger, ts *TailService, wg *sync.WaitGroup) *Hub

NewHub creates hub for client services

func (*Hub) Close

func (h *Hub) Close()

Close closes message processing

func (*Hub) Run

func (h *Hub) Run()

Run processes hub messages

type InMessage

type InMessage struct {
	Type    string `json:"type"`
	Channel string `json:"channel,omitempty"`
}

InMessage holds incoming client request

type IndexItemAttr

type IndexItemAttr struct {
	ModTime time.Time `json:"mtime"`
	Size    int64     `json:"size"`
}

IndexItemAttr holds File (index item) Attrs

type IndexItemAttrStore

type IndexItemAttrStore map[string]*IndexItemAttr

IndexItemAttrStore holds all index items

type IndexItemEvent

type IndexItemEvent struct {
	ModTime time.Time `json:"mtime"`
	Size    int64     `json:"size"`
	Name    string    `json:"name"`
	Deleted bool      `json:"deleted,omitempty"`
}

IndexItemEvent holds messages from indexer

type IndexMessage

type IndexMessage struct {
	Type  string         `json:"type"`
	Data  IndexItemEvent `json:"data"`
	Error string         `json:"error,omitempty"`
}

IndexMessage holds outgoing message item for file index

type Message

type Message struct {
	Client  *Client
	Message []byte
}

Message holds received message and sender

type Service

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

Service holds WebTail service

func New

func New(log logr.Logger, cfg *Config) (*Service, error)

New creates WebTail service

func (*Service) Close

func (wt *Service) Close()

Close stops a message hub

func (*Service) Run

func (wt *Service) Run()

Run runs a message hub

func (*Service) ServeHTTP

func (wt *Service) ServeHTTP(w http.ResponseWriter, r *http.Request)

Handle handles websocket requests from the peer

type StatsMessage

type StatsMessage struct {
	Type string            `json:"type"`
	Data map[string]uint64 `json:"data,omitempty"`
}

StatsMessage holds outgoing app stats

type TailAttr

type TailAttr struct {
	// Store for last Config.Lines lines
	Buffer [][]byte

	// Quit worker process
	Quit chan struct{}

	// Skip 1st line when read file not from start
	IsHeadTrimmed bool
}

TailAttr holds tail worker attributes

type TailMessage

type TailMessage struct {
	Type    string `json:"type"`
	Channel string `json:"channel,omitempty"`
	Data    string `json:"data,omitempty"`
}

TailMessage holds outgoing file tail row

type TailService

type TailService struct {
	Config *Config
	// contains filtered or unexported fields
}

TailService holds Worker hub operations

func NewTailService

func NewTailService(logger logr.Logger, cfg *Config) (*TailService, error)

NewTailService creates tailer service

func (*TailService) ChannelExists

func (ts *TailService) ChannelExists(channel string) bool

ChannelExists checks if channel allowed to attach

func (*TailService) IndexItem

func (ts *TailService) IndexItem(key string) *IndexItemAttr

IndexItem returns index item

func (*TailService) IndexKeys

func (ts *TailService) IndexKeys() []string

IndexKeys returns sorted index keys

func (*TailService) IndexUpdate

func (ts *TailService) IndexUpdate(msg *IndexItemEvent)

IndexUpdate updates TailService index item

func (*TailService) IndexerRun

func (ts *TailService) IndexerRun(out chan *IndexItemEvent, wg *sync.WaitGroup)

IndexerRun runs indexer

func (*TailService) SetTrace

func (ts *TailService) SetTrace(mode string)

SetTrace turns on/off logging of incoming workers messages

func (*TailService) TailerAppend

func (ts *TailService) TailerAppend(channel string, data []byte) bool

TailerAppend adds a line into worker buffer

func (*TailService) TailerBuffer

func (ts *TailService) TailerBuffer(channel string) [][]byte

TailerBuffer returns worker buffer

func (*TailService) TailerRun

func (ts *TailService) TailerRun(channel string, out chan *TailMessage, readyChan chan struct{}, wg *sync.WaitGroup) error

TailerRun creates and runs tail worker

func (*TailService) TraceEnabled

func (ts *TailService) TraceEnabled() bool

TraceEnabled returns trace state

func (*TailService) WorkerExists

func (ts *TailService) WorkerExists(channel string) bool

WorkerExists checks if worker already registered

func (*TailService) WorkerStop

func (ts *TailService) WorkerStop(channel string)

WorkerStop stops worker (tailer or indexer)

type TraceMessage

type TraceMessage struct {
	Type    string `json:"type"`
	Enabled bool   `json:"enabled"`
}

TraceMessage holds outgoing trace state