/usr/share/gocode/src/github.com/coreos/go-systemd/sdjournal/read.go is in golang-github-coreos-go-systemd-dev 14-1.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 | // Copyright 2015 RedHat, Inc.
// Copyright 2015 CoreOS, Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package sdjournal
import (
"errors"
"fmt"
"io"
"log"
"strings"
"time"
)
var (
ErrExpired = errors.New("Timeout expired")
)
// JournalReaderConfig represents options to drive the behavior of a JournalReader.
type JournalReaderConfig struct {
// The Since, NumFromTail and Cursor options are mutually exclusive and
// determine where the reading begins within the journal. The order in which
// options are written is exactly the order of precedence.
Since time.Duration // start relative to a Duration from now
NumFromTail uint64 // start relative to the tail
Cursor string // start relative to the cursor
// Show only journal entries whose fields match the supplied values. If
// the array is empty, entries will not be filtered.
Matches []Match
// If not empty, the journal instance will point to a journal residing
// in this directory. The supplied path may be relative or absolute.
Path string
}
// JournalReader is an io.ReadCloser which provides a simple interface for iterating through the
// systemd journal. A JournalReader is not safe for concurrent use by multiple goroutines.
type JournalReader struct {
journal *Journal
msgReader *strings.Reader
}
// NewJournalReader creates a new JournalReader with configuration options that are similar to the
// systemd journalctl tool's iteration and filtering features.
func NewJournalReader(config JournalReaderConfig) (*JournalReader, error) {
r := &JournalReader{}
// Open the journal
var err error
if config.Path != "" {
r.journal, err = NewJournalFromDir(config.Path)
} else {
r.journal, err = NewJournal()
}
if err != nil {
return nil, err
}
// Add any supplied matches
for _, m := range config.Matches {
r.journal.AddMatch(m.String())
}
// Set the start position based on options
if config.Since != 0 {
// Start based on a relative time
start := time.Now().Add(config.Since)
if err := r.journal.SeekRealtimeUsec(uint64(start.UnixNano() / 1000)); err != nil {
return nil, err
}
} else if config.NumFromTail != 0 {
// Start based on a number of lines before the tail
if err := r.journal.SeekTail(); err != nil {
return nil, err
}
// Move the read pointer into position near the tail. Go one further than
// the option so that the initial cursor advancement positions us at the
// correct starting point.
skip, err := r.journal.PreviousSkip(config.NumFromTail + 1)
if err != nil {
return nil, err
}
// If we skipped fewer lines than expected, we have reached journal start.
// Thus, we seek to head so that next invocation can read the first line.
if skip != config.NumFromTail+1 {
if err := r.journal.SeekHead(); err != nil {
return nil, err
}
}
} else if config.Cursor != "" {
// Start based on a custom cursor
if err := r.journal.SeekCursor(config.Cursor); err != nil {
return nil, err
}
}
return r, nil
}
// Read reads entries from the journal. Read follows the Reader interface so
// it must be able to read a specific amount of bytes. Journald on the other
// hand only allows us to read full entries of arbitrary size (without byte
// granularity). JournalReader is therefore internally buffering entries that
// don't fit in the read buffer. Callers should keep calling until 0 and/or an
// error is returned.
func (r *JournalReader) Read(b []byte) (int, error) {
var err error
if r.msgReader == nil {
var c uint64
// Advance the journal cursor. It has to be called at least one time
// before reading
c, err = r.journal.Next()
// An unexpected error
if err != nil {
return 0, err
}
// EOF detection
if c == 0 {
return 0, io.EOF
}
// Build a message
var msg string
msg, err = r.buildMessage()
if err != nil {
return 0, err
}
r.msgReader = strings.NewReader(msg)
}
// Copy and return the message
var sz int
sz, err = r.msgReader.Read(b)
if err == io.EOF {
// The current entry has been fully read. Don't propagate this
// EOF, so the next entry can be read at the next Read()
// iteration.
r.msgReader = nil
return sz, nil
}
if err != nil {
return sz, err
}
if r.msgReader.Len() == 0 {
r.msgReader = nil
}
return sz, nil
}
// Close closes the JournalReader's handle to the journal.
func (r *JournalReader) Close() error {
return r.journal.Close()
}
// Rewind attempts to rewind the JournalReader to the first entry.
func (r *JournalReader) Rewind() error {
r.msgReader = nil
return r.journal.SeekHead()
}
// Follow synchronously follows the JournalReader, writing each new journal entry to writer. The
// follow will continue until a single time.Time is received on the until channel.
func (r *JournalReader) Follow(until <-chan time.Time, writer io.Writer) (err error) {
// Process journal entries and events. Entries are flushed until the tail or
// timeout is reached, and then we wait for new events or the timeout.
var msg = make([]byte, 64*1<<(10))
process:
for {
c, err := r.Read(msg)
if err != nil && err != io.EOF {
break process
}
select {
case <-until:
return ErrExpired
default:
if c > 0 {
if _, err = writer.Write(msg[:c]); err != nil {
break process
}
continue process
}
}
// We're at the tail, so wait for new events or time out.
// Holds journal events to process. Tightly bounded for now unless there's a
// reason to unblock the journal watch routine more quickly.
events := make(chan int, 1)
pollDone := make(chan bool, 1)
go func() {
for {
select {
case <-pollDone:
return
default:
events <- r.journal.Wait(time.Duration(1) * time.Second)
}
}
}()
select {
case <-until:
pollDone <- true
return ErrExpired
case e := <-events:
pollDone <- true
switch e {
case SD_JOURNAL_NOP, SD_JOURNAL_APPEND, SD_JOURNAL_INVALIDATE:
// TODO: need to account for any of these?
default:
log.Printf("Received unknown event: %d\n", e)
}
continue process
}
}
return
}
// buildMessage returns a string representing the current journal entry in a simple format which
// includes the entry timestamp and MESSAGE field.
func (r *JournalReader) buildMessage() (string, error) {
var msg string
var usec uint64
var err error
if msg, err = r.journal.GetData("MESSAGE"); err != nil {
return "", err
}
if usec, err = r.journal.GetRealtimeUsec(); err != nil {
return "", err
}
timestamp := time.Unix(0, int64(usec)*int64(time.Microsecond))
return fmt.Sprintf("%s %s\n", timestamp, msg), nil
}
|