-
Notifications
You must be signed in to change notification settings - Fork 173
/
log_reader.go
150 lines (132 loc) · 3.47 KB
/
log_reader.go
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
// Copyright 2017 VMware, Inc. All Rights Reserved.
//
// 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 iolog
import (
"encoding/base64"
"encoding/binary"
"io"
"time"
)
const (
// RFC3339NanoFixed is Docker's version of RFC339Nano that pads the
// nanoseconds with zeros to ensure that the timestamps are aligned in the logs.
RFC3339NanoFixed = "2006-01-02T15:04:05.000000000Z07:00"
)
// LogReader reads containerVM entries from a stream and decodes them into
// their original form
type LogReader struct {
io.ReadCloser
prev []byte
ts bool
}
// Header describes the header of a containerVM log entry
type Header struct {
Timestamp time.Time
Size int
Stream int
}
// Entry describes a containerVM log entry
type Entry struct {
Header
Message []byte
}
// NewLogReader wraps an io.ReadCloser in a LogReader.
func NewLogReader(r io.ReadCloser, ts bool) *LogReader {
return &LogReader{
ReadCloser: r,
ts: ts}
}
// ParseLogEntry parses data from an io.Reader into a Entry
func ParseLogEntry(r io.Reader) (*Entry, error) {
var (
err error
n, w int
ts time.Time
)
ehdr := make([]byte, encodedHeaderLengthBytes)
// read a header
n = 0
for n < encodedHeaderLengthBytes {
w, err = r.Read(ehdr[n:])
n += w
if err != nil {
return nil, err
}
}
// decode base64 header
hdr, err := base64.StdEncoding.DecodeString(string(ehdr))
if err != nil {
return nil, err
}
// parse header
ts = time.Unix(0, int64(binary.LittleEndian.Uint64(hdr[:8])))
s := binary.LittleEndian.Uint16(hdr[8:10])
// stream := int((s&streamFlag) >> 3)
size := int(s >> 4)
entry := &Entry{
Header: Header{
Timestamp: ts,
Size: size,
},
}
// read the associated entry
msg := make([]byte, size)
n = 0
for n < size {
w, err = r.Read(msg[n:])
n += w
if err != nil {
if err != io.EOF {
// only return if not EOF as we may actually have some bytes to copy
return nil, err
}
break
}
}
entry.Message = msg
return entry, nil
}
// Read reads a 10 byte header and decodes it into the timestamp, stream and
// size of an entry. It uses the size to read the next set of bytes as the
// message, and then copies the message into the supplied buffer, saving
// what will not fit in the buffer for the next call to Read.
func (lr *LogReader) Read(p []byte) (int, error) {
var (
entry *Entry
err error
)
msg := lr.prev
partial := true // treat msg as a partial entry until we verify otherwise
if msg == nil {
// we know msg is not a partial entry as we had no bytes left from the previous call
partial = false
entry, err = ParseLogEntry(lr.ReadCloser)
if err != nil {
return 0, err
}
msg = entry.Message
}
lr.prev = nil
if len(p) < len(msg) {
// copy what we can and save the rest for the next call
lr.prev = msg[len(p):]
msg = msg[:len(p)]
}
// add timestamp if enabled
if lr.ts && !partial {
msg = append([]byte(entry.Timestamp.Format(RFC3339NanoFixed)+" "), msg...)
}
// write the log message
return copy(p, msg), err
}