path: root/dashboard/log.go

// Copyright 2018 The go-ethereum Authors
// This file is part of the go-ethereum library.
//
// The go-ethereum library is free software: you can redistribute it and/or modify
// it under the terms of the GNU Lesser General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// The go-ethereum library is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Lesser General Public License for more details.
//
// You should have received a copy of the GNU Lesser General Public License
// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.

package dashboard

import (
    "bytes"
    "encoding/json"
    "io/ioutil"
    "os"
    "path/filepath"
    "regexp"
    "sort"
    "time"

    "github.com/mohae/deepcopy"
    "github.com/rjeczalik/notify"
    "github.com/tangerine-network/go-tangerine/log"
)

var emptyChunk = json.RawMessage("[]")

// prepLogs creates a JSON array from the given log record buffer.
// Returns the prepared array and the position of the last '\n'
// character in the original buffer, or -1 if it doesn't contain any.
func prepLogs(buf []byte) (json.RawMessage, int) {
    b := make(json.RawMessage, 1, len(buf)+1)
    b[0] = '['
    b = append(b, buf...)
    last := -1
    for i := 1; i < len(b); i++ {
        if b[i] == '\n' {
            b[i] = ','
            last = i
        }
    }
    if last < 0 {
        return emptyChunk, -1
    }
    b[last] = ']'
    return b[:last+1], last - 1
}

// handleLogRequest searches for the log file specified by the timestamp of the
// request, creates a JSON array out of it and sends it to the requesting client.
func (db *Dashboard) handleLogRequest(r *LogsRequest, c *client) {
    files, err := ioutil.ReadDir(db.logdir)
    if err != nil {
        log.Warn("Failed to open logdir", "path", db.logdir, "err", err)
        return
    }
    re := regexp.MustCompile(`\.log$`)
    fileNames := make([]string, 0, len(files))
    for _, f := range files {
        if f.Mode().IsRegular() && re.MatchString(f.Name()) {
            fileNames = append(fileNames, f.Name())
        }
    }
    if len(fileNames) < 1 {
        log.Warn("No log files in logdir", "path", db.logdir)
        return
    }
    idx := sort.Search(len(fileNames), func(idx int) bool {
        // Returns the smallest index such as fileNames[idx] >= r.Name,
        // if there is no such index, returns n.
        return fileNames[idx] >= r.Name
    })

    switch {
    case idx < 0:
        return
    case idx == 0 && r.Past:
        return
    case idx >= len(fileNames):
        return
    case r.Past:
        idx--
    case idx == len(fileNames)-1 && fileNames[idx] == r.Name:
        return
    case idx == len(fileNames)-1 || (idx == len(fileNames)-2 && fileNames[idx] == r.Name):
        // The last file is continuously updated, and its chunks are streamed,
        // so in order to avoid log record duplication on the client side, it is
        // handled differently. Its actual content is always saved in the history.
        db.lock.Lock()
        if db.history.Logs != nil {
            c.msg <- &Message{
                Logs: db.history.Logs,
            }
        }
        db.lock.Unlock()
        return
    case fileNames[idx] == r.Name:
        idx++
    }

    path := filepath.Join(db.logdir, fileNames[idx])
    var buf []byte
    if buf, err = ioutil.ReadFile(path); err != nil {
        log.Warn("Failed to read file", "path", path, "err", err)
        return
    }
    chunk, end := prepLogs(buf)
    if end < 0 {
        log.Warn("The file doesn't contain valid logs", "path", path)
        return
    }
    c.msg <- &Message{
        Logs: &LogsMessage{
            Source: &LogFile{
                Name: fileNames[idx],
                Last: r.Past && idx == 0,
            },
            Chunk: chunk,
        },
    }
}

// streamLogs watches the file system, and when the logger writes
// the new log records into the files, picks them up, then makes
// JSON array out of them and sends them to the clients.
func (db *Dashboard) streamLogs() {
    defer db.wg.Done()
    var (
        err  error
        errc chan error
    )
    defer func() {
        if errc == nil {
            errc = <-db.quit
        }
        errc <- err
    }()

    files, err := ioutil.ReadDir(db.logdir)
    if err != nil {
        log.Warn("Failed to open logdir", "path", db.logdir, "err", err)
        return
    }
    var (
        opened *os.File // File descriptor for the opened active log file.
        buf    []byte   // Contains the recently written log chunks, which are not sent to the clients yet.
    )

    // The log records are always written into the last file in alphabetical order, because of the timestamp.
    re := regexp.MustCompile(`\.log$`)
    i := len(files) - 1
    for i >= 0 && (!files[i].Mode().IsRegular() || !re.MatchString(files[i].Name())) {
        i--
    }
    if i < 0 {
        log.Warn("No log files in logdir", "path", db.logdir)
        return
    }
    if opened, err = os.OpenFile(filepath.Join(db.logdir, files[i].Name()), os.O_RDONLY, 0600); err != nil {
        log.Warn("Failed to open file", "name", files[i].Name(), "err", err)
        return
    }
    defer opened.Close() // Close the lastly opened file.
    fi, err := opened.Stat()
    if err != nil {
        log.Warn("Problem with file", "name", opened.Name(), "err", err)
        return
    }
    db.lock.Lock()
    db.history.Logs = &LogsMessage{
        Source: &LogFile{
            Name: fi.Name(),
            Last: true,
        },
        Chunk: emptyChunk,
    }
    db.lock.Unlock()

    watcher := make(chan notify.EventInfo, 10)
    if err := notify.Watch(db.logdir, watcher, notify.Create); err != nil {
        log.Warn("Failed to create file system watcher", "err", err)
        return
    }
    defer notify.Stop(watcher)

    ticker := time.NewTicker(db.config.Refresh)
    defer ticker.Stop()

loop:
    for err == nil || errc == nil {
        select {
        case event := <-watcher:
            // Make sure that new log file was created.
            if !re.Match([]byte(event.Path())) {
                break
            }
            if opened == nil {
                log.Warn("The last log file is not opened")
                break loop
            }
            // The new log file's name is always greater,
            // because it is created using the actual log record's time.
            if opened.Name() >= event.Path() {
                break
            }
            // Read the rest of the previously opened file.
            chunk, err := ioutil.ReadAll(opened)
            if err != nil {
                log.Warn("Failed to read file", "name", opened.Name(), "err", err)
                break loop
            }
            buf = append(buf, chunk...)
            opened.Close()

            if chunk, last := prepLogs(buf); last >= 0 {
                // Send the rest of the previously opened file.
                db.sendToAll(&Message{
                    Logs: &LogsMessage{
                        Chunk: chunk,
                    },
                })
            }
            if opened, err = os.OpenFile(event.Path(), os.O_RDONLY, 0644); err != nil {
                log.Warn("Failed to open file", "name", event.Path(), "err", err)
                break loop
            }
            buf = buf[:0]

            // Change the last file in the history.
            fi, err := opened.Stat()
            if err != nil {
                log.Warn("Problem with file", "name", opened.Name(), "err", err)
                break loop
            }
            db.lock.Lock()
            db.history.Logs.Source.Name = fi.Name()
            db.history.Logs.Chunk = emptyChunk
            db.lock.Unlock()
        case <-ticker.C: // Send log updates to the client.
            if opened == nil {
                log.Warn("The last log file is not opened")
                break loop
            }
            // Read the new logs created since the last read.
            chunk, err := ioutil.ReadAll(opened)
            if err != nil {
                log.Warn("Failed to read file", "name", opened.Name(), "err", err)
                break loop
            }
            b := append(buf, chunk...)

            chunk, last := prepLogs(b)
            if last < 0 {
                break
            }
            // Only keep the invalid part of the buffer, which can be valid after the next read.
            buf = b[last+1:]

            var l *LogsMessage
            // Update the history.
            db.lock.Lock()
            if bytes.Equal(db.history.Logs.Chunk, emptyChunk) {
                db.history.Logs.Chunk = chunk
                l = deepcopy.Copy(db.history.Logs).(*LogsMessage)
            } else {
                b = make([]byte, len(db.history.Logs.Chunk)+len(chunk)-1)
                copy(b, db.history.Logs.Chunk)
                b[len(db.history.Logs.Chunk)-1] = ','
                copy(b[len(db.history.Logs.Chunk):], chunk[1:])
                db.history.Logs.Chunk = b
                l = &LogsMessage{Chunk: chunk}
            }
            db.lock.Unlock()

            db.sendToAll(&Message{Logs: l})
        case errc = <-db.quit:
            break loop
        }
    }
}