1. docker
  2. Commit e0ad6d045c752e0523d8591b235ec2db32bc71fc
Browse code

Merge pull request #37092 from cpuguy83/local_logger

Add "local" log driver

Sebastiaan van Stijn authored on 2018/08/20 15:01:41
Showing 17 changed files
api/types/plugins/logdriver/entry.pb.go
History View file @ e0ad6d0
... ... 
@@ -10,6 +10,7 @@
10 10
11 11 
 	It has these top-level messages:
12 12 
 		LogEntry
13 
+		PartialLogEntryMetadata
13 14 
 */
14 15 
 package logdriver
15 16
... ... 
@@ -31,10 +32,11 @@ var _ = math.Inf
31 31 
 const _ = proto.GoGoProtoPackageIsVersion2 // please upgrade the proto package
32 32
33 33 
 type LogEntry struct {
34 
-	Source   string `protobuf:"bytes,1,opt,name=source,proto3" json:"source,omitempty"`
35 
-	TimeNano int64  `protobuf:"varint,2,opt,name=time_nano,json=timeNano,proto3" json:"time_nano,omitempty"`
36 
-	Line     []byte `protobuf:"bytes,3,opt,name=line,proto3" json:"line,omitempty"`
37 
-	Partial  bool   `protobuf:"varint,4,opt,name=partial,proto3" json:"partial,omitempty"`
34 
+	Source             string                   `protobuf:"bytes,1,opt,name=source,proto3" json:"source,omitempty"`
35 
+	TimeNano           int64                    `protobuf:"varint,2,opt,name=time_nano,json=timeNano,proto3" json:"time_nano,omitempty"`
36 
+	Line               []byte                   `protobuf:"bytes,3,opt,name=line,proto3" json:"line,omitempty"`
37 
+	Partial            bool                     `protobuf:"varint,4,opt,name=partial,proto3" json:"partial,omitempty"`
38 
+	PartialLogMetadata *PartialLogEntryMetadata `protobuf:"bytes,5,opt,name=partial_log_metadata,json=partialLogMetadata" json:"partial_log_metadata,omitempty"`
38 39 
 }
39 40
40 41 
 func (m *LogEntry) Reset()                    { *m = LogEntry{} }
... ... 
@@ -70,8 +72,48 @@ func (m *LogEntry) GetPartial() bool {
70 70 
 	return false
71 71 
 }
72 72
73 
+func (m *LogEntry) GetPartialLogMetadata() *PartialLogEntryMetadata {
74 
+	if m != nil {
75 
+		return m.PartialLogMetadata
76 
+	}
77 
+	return nil
78 
+}
79 
+
80 
+type PartialLogEntryMetadata struct {
81 
+	Last    bool   `protobuf:"varint,1,opt,name=last,proto3" json:"last,omitempty"`
82 
+	Id      string `protobuf:"bytes,2,opt,name=id,proto3" json:"id,omitempty"`
83 
+	Ordinal int32  `protobuf:"varint,3,opt,name=ordinal,proto3" json:"ordinal,omitempty"`
84 
+}
85 
+
86 
+func (m *PartialLogEntryMetadata) Reset()                    { *m = PartialLogEntryMetadata{} }
87 
+func (m *PartialLogEntryMetadata) String() string            { return proto.CompactTextString(m) }
88 
+func (*PartialLogEntryMetadata) ProtoMessage()               {}
89 
+func (*PartialLogEntryMetadata) Descriptor() ([]byte, []int) { return fileDescriptorEntry, []int{1} }
90 
+
91 
+func (m *PartialLogEntryMetadata) GetLast() bool {
92 
+	if m != nil {
93 
+		return m.Last
94 
+	}
95 
+	return false
96 
+}
97 
+
98 
+func (m *PartialLogEntryMetadata) GetId() string {
99 
+	if m != nil {
100 
+		return m.Id
101 
+	}
102 
+	return ""
103 
+}
104 
+
105 
+func (m *PartialLogEntryMetadata) GetOrdinal() int32 {
106 
+	if m != nil {
107 
+		return m.Ordinal
108 
+	}
109 
+	return 0
110 
+}
111 
+
73 112 
 func init() {
74 113 
 	proto.RegisterType((*LogEntry)(nil), "LogEntry")
114 
+	proto.RegisterType((*PartialLogEntryMetadata)(nil), "PartialLogEntryMetadata")
75 115 
 }
76 116 
 func (m *LogEntry) Marshal() (dAtA []byte, err error) {
77 117 
 	size := m.Size()
... ... 
@@ -115,6 +157,55 @@ func (m *LogEntry) MarshalTo(dAtA []byte) (int, error) {
115 115 
 		}
116 116 
 		i++
117 117 
 	}
118 
+	if m.PartialLogMetadata != nil {
119 
+		dAtA[i] = 0x2a
120 
+		i++
121 
+		i = encodeVarintEntry(dAtA, i, uint64(m.PartialLogMetadata.Size()))
122 
+		n1, err := m.PartialLogMetadata.MarshalTo(dAtA[i:])
123 
+		if err != nil {
124 
+			return 0, err
125 
+		}
126 
+		i += n1
127 
+	}
128 
+	return i, nil
129 
+}
130 
+
131 
+func (m *PartialLogEntryMetadata) Marshal() (dAtA []byte, err error) {
132 
+	size := m.Size()
133 
+	dAtA = make([]byte, size)
134 
+	n, err := m.MarshalTo(dAtA)
135 
+	if err != nil {
136 
+		return nil, err
137 
+	}
138 
+	return dAtA[:n], nil
139 
+}
140 
+
141 
+func (m *PartialLogEntryMetadata) MarshalTo(dAtA []byte) (int, error) {
142 
+	var i int
143 
+	_ = i
144 
+	var l int
145 
+	_ = l
146 
+	if m.Last {
147 
+		dAtA[i] = 0x8
148 
+		i++
149 
+		if m.Last {
150 
+			dAtA[i] = 1
151 
+		} else {
152 
+			dAtA[i] = 0
153 
+		}
154 
+		i++
155 
+	}
156 
+	if len(m.Id) > 0 {
157 
+		dAtA[i] = 0x12
158 
+		i++
159 
+		i = encodeVarintEntry(dAtA, i, uint64(len(m.Id)))
160 
+		i += copy(dAtA[i:], m.Id)
161 
+	}
162 
+	if m.Ordinal != 0 {
163 
+		dAtA[i] = 0x18
164 
+		i++
165 
+		i = encodeVarintEntry(dAtA, i, uint64(m.Ordinal))
166 
+	}
118 167 
 	return i, nil
119 168 
 }
120 169
... ... 
@@ -162,6 +253,26 @@ func (m *LogEntry) Size() (n int) {
162 162 
 	if m.Partial {
163 163 
 		n += 2
164 164 
 	}
165 
+	if m.PartialLogMetadata != nil {
166 
+		l = m.PartialLogMetadata.Size()
167 
+		n += 1 + l + sovEntry(uint64(l))
168 
+	}
169 
+	return n
170 
+}
171 
+
172 
+func (m *PartialLogEntryMetadata) Size() (n int) {
173 
+	var l int
174 
+	_ = l
175 
+	if m.Last {
176 
+		n += 2
177 
+	}
178 
+	l = len(m.Id)
179 
+	if l > 0 {
180 
+		n += 1 + l + sovEntry(uint64(l))
181 
+	}
182 
+	if m.Ordinal != 0 {
183 
+		n += 1 + sovEntry(uint64(m.Ordinal))
184 
+	}
165 185 
 	return n
166 186 
 }
167 187
... ... 
@@ -306,6 +417,157 @@ func (m *LogEntry) Unmarshal(dAtA []byte) error {
306 306 
 				}
307 307 
 			}
308 308 
 			m.Partial = bool(v != 0)
309 
+		case 5:
310 
+			if wireType != 2 {
311 
+				return fmt.Errorf("proto: wrong wireType = %d for field PartialLogMetadata", wireType)
312 
+			}
313 
+			var msglen int
314 
+			for shift := uint(0); ; shift += 7 {
315 
+				if shift >= 64 {
316 
+					return ErrIntOverflowEntry
317 
+				}
318 
+				if iNdEx >= l {
319 
+					return io.ErrUnexpectedEOF
320 
+				}
321 
+				b := dAtA[iNdEx]
322 
+				iNdEx++
323 
+				msglen |= (int(b) & 0x7F) << shift
324 
+				if b < 0x80 {
325 
+					break
326 
+				}
327 
+			}
328 
+			if msglen < 0 {
329 
+				return ErrInvalidLengthEntry
330 
+			}
331 
+			postIndex := iNdEx + msglen
332 
+			if postIndex > l {
333 
+				return io.ErrUnexpectedEOF
334 
+			}
335 
+			if m.PartialLogMetadata == nil {
336 
+				m.PartialLogMetadata = &PartialLogEntryMetadata{}
337 
+			}
338 
+			if err := m.PartialLogMetadata.Unmarshal(dAtA[iNdEx:postIndex]); err != nil {
339 
+				return err
340 
+			}
341 
+			iNdEx = postIndex
342 
+		default:
343 
+			iNdEx = preIndex
344 
+			skippy, err := skipEntry(dAtA[iNdEx:])
345 
+			if err != nil {
346 
+				return err
347 
+			}
348 
+			if skippy < 0 {
349 
+				return ErrInvalidLengthEntry
350 
+			}
351 
+			if (iNdEx + skippy) > l {
352 
+				return io.ErrUnexpectedEOF
353 
+			}
354 
+			iNdEx += skippy
355 
+		}
356 
+	}
357 
+
358 
+	if iNdEx > l {
359 
+		return io.ErrUnexpectedEOF
360 
+	}
361 
+	return nil
362 
+}
363 
+func (m *PartialLogEntryMetadata) Unmarshal(dAtA []byte) error {
364 
+	l := len(dAtA)
365 
+	iNdEx := 0
366 
+	for iNdEx < l {
367 
+		preIndex := iNdEx
368 
+		var wire uint64
369 
+		for shift := uint(0); ; shift += 7 {
370 
+			if shift >= 64 {
371 
+				return ErrIntOverflowEntry
372 
+			}
373 
+			if iNdEx >= l {
374 
+				return io.ErrUnexpectedEOF
375 
+			}
376 
+			b := dAtA[iNdEx]
377 
+			iNdEx++
378 
+			wire |= (uint64(b) & 0x7F) << shift
379 
+			if b < 0x80 {
380 
+				break
381 
+			}
382 
+		}
383 
+		fieldNum := int32(wire >> 3)
384 
+		wireType := int(wire & 0x7)
385 
+		if wireType == 4 {
386 
+			return fmt.Errorf("proto: PartialLogEntryMetadata: wiretype end group for non-group")
387 
+		}
388 
+		if fieldNum <= 0 {
389 
+			return fmt.Errorf("proto: PartialLogEntryMetadata: illegal tag %d (wire type %d)", fieldNum, wire)
390 
+		}
391 
+		switch fieldNum {
392 
+		case 1:
393 
+			if wireType != 0 {
394 
+				return fmt.Errorf("proto: wrong wireType = %d for field Last", wireType)
395 
+			}
396 
+			var v int
397 
+			for shift := uint(0); ; shift += 7 {
398 
+				if shift >= 64 {
399 
+					return ErrIntOverflowEntry
400 
+				}
401 
+				if iNdEx >= l {
402 
+					return io.ErrUnexpectedEOF
403 
+				}
404 
+				b := dAtA[iNdEx]
405 
+				iNdEx++
406 
+				v |= (int(b) & 0x7F) << shift
407 
+				if b < 0x80 {
408 
+					break
409 
+				}
410 
+			}
411 
+			m.Last = bool(v != 0)
412 
+		case 2:
413 
+			if wireType != 2 {
414 
+				return fmt.Errorf("proto: wrong wireType = %d for field Id", wireType)
415 
+			}
416 
+			var stringLen uint64
417 
+			for shift := uint(0); ; shift += 7 {
418 
+				if shift >= 64 {
419 
+					return ErrIntOverflowEntry
420 
+				}
421 
+				if iNdEx >= l {
422 
+					return io.ErrUnexpectedEOF
423 
+				}
424 
+				b := dAtA[iNdEx]
425 
+				iNdEx++
426 
+				stringLen |= (uint64(b) & 0x7F) << shift
427 
+				if b < 0x80 {
428 
+					break
429 
+				}
430 
+			}
431 
+			intStringLen := int(stringLen)
432 
+			if intStringLen < 0 {
433 
+				return ErrInvalidLengthEntry
434 
+			}
435 
+			postIndex := iNdEx + intStringLen
436 
+			if postIndex > l {
437 
+				return io.ErrUnexpectedEOF
438 
+			}
439 
+			m.Id = string(dAtA[iNdEx:postIndex])
440 
+			iNdEx = postIndex
441 
+		case 3:
442 
+			if wireType != 0 {
443 
+				return fmt.Errorf("proto: wrong wireType = %d for field Ordinal", wireType)
444 
+			}
445 
+			m.Ordinal = 0
446 
+			for shift := uint(0); ; shift += 7 {
447 
+				if shift >= 64 {
448 
+					return ErrIntOverflowEntry
449 
+				}
450 
+				if iNdEx >= l {
451 
+					return io.ErrUnexpectedEOF
452 
+				}
453 
+				b := dAtA[iNdEx]
454 
+				iNdEx++
455 
+				m.Ordinal |= (int32(b) & 0x7F) << shift
456 
+				if b < 0x80 {
457 
+					break
458 
+				}
459 
+			}
309 460 
 		default:
310 461 
 			iNdEx = preIndex
311 462 
 			skippy, err := skipEntry(dAtA[iNdEx:])
... ... 
@@ -435,15 +697,20 @@ var (
435 435 
 func init() { proto.RegisterFile("entry.proto", fileDescriptorEntry) }
436 436
437 437 
 var fileDescriptorEntry = []byte{
438 
-	// 149 bytes of a gzipped FileDescriptorProto
439 
-	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x09, 0x6e, 0x88, 0x02, 0xff, 0xe2, 0xe2, 0x4e, 0xcd, 0x2b, 0x29,
440 
-	0xaa, 0xd4, 0x2b, 0x28, 0xca, 0x2f, 0xc9, 0x57, 0xca, 0xe5, 0xe2, 0xf0, 0xc9, 0x4f, 0x77, 0x05,
441 
-	0x89, 0x08, 0x89, 0x71, 0xb1, 0x15, 0xe7, 0x97, 0x16, 0x25, 0xa7, 0x4a, 0x30, 0x2a, 0x30, 0x6a,
442 
-	0x70, 0x06, 0x41, 0x79, 0x42, 0xd2, 0x5c, 0x9c, 0x25, 0x99, 0xb9, 0xa9, 0xf1, 0x79, 0x89, 0x79,
443 
-	0xf9, 0x12, 0x4c, 0x0a, 0x8c, 0x1a, 0xcc, 0x41, 0x1c, 0x20, 0x01, 0xbf, 0xc4, 0xbc, 0x7c, 0x21,
444 
-	0x21, 0x2e, 0x96, 0x9c, 0xcc, 0xbc, 0x54, 0x09, 0x66, 0x05, 0x46, 0x0d, 0x9e, 0x20, 0x30, 0x5b,
445 
-	0x48, 0x82, 0x8b, 0xbd, 0x20, 0xb1, 0xa8, 0x24, 0x33, 0x31, 0x47, 0x82, 0x45, 0x81, 0x51, 0x83,
446 
-	0x23, 0x08, 0xc6, 0x75, 0xe2, 0x39, 0xf1, 0x48, 0x8e, 0xf1, 0xc2, 0x23, 0x39, 0xc6, 0x07, 0x8f,
447 
-	0xe4, 0x18, 0x93, 0xd8, 0xc0, 0x6e, 0x30, 0x06, 0x04, 0x00, 0x00, 0xff, 0xff, 0x2d, 0x24, 0x5a,
448 
-	0xd4, 0x92, 0x00, 0x00, 0x00,
438 
+	// 237 bytes of a gzipped FileDescriptorProto
439 
+	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x09, 0x6e, 0x88, 0x02, 0xff, 0x74, 0x90, 0xbd, 0x4a, 0x04, 0x31,
440 
+	0x14, 0x85, 0xb9, 0xb3, 0x3f, 0xce, 0xdc, 0x5d, 0x2c, 0x82, 0x68, 0x40, 0x18, 0xc2, 0x56, 0xa9,
441 
+	0xb6, 0xd0, 0x37, 0x10, 0x6c, 0x44, 0x45, 0xd2, 0x58, 0x0e, 0x57, 0x27, 0x2c, 0x81, 0xd9, 0xdc,
442 
+	0x21, 0x13, 0x0b, 0x1f, 0xcd, 0x37, 0xb0, 0xf4, 0x11, 0x64, 0x9e, 0x44, 0x26, 0x4e, 0xec, 0xec,
443 
+	0xce, 0x39, 0x5f, 0x8a, 0x2f, 0x17, 0x37, 0xd6, 0xc7, 0xf0, 0xbe, 0xef, 0x03, 0x47, 0xde, 0x7d,
444 
+	0x00, 0x96, 0xf7, 0x7c, 0xb8, 0x9d, 0x26, 0x71, 0x8e, 0xeb, 0x81, 0xdf, 0xc2, 0xab, 0x95, 0xa0,
445 
+	0x40, 0x57, 0x66, 0x6e, 0xe2, 0x12, 0xab, 0xe8, 0x8e, 0xb6, 0xf1, 0xe4, 0x59, 0x16, 0x0a, 0xf4,
446 
+	0xc2, 0x94, 0xd3, 0xf0, 0x48, 0x9e, 0x85, 0xc0, 0x65, 0xe7, 0xbc, 0x95, 0x0b, 0x05, 0x7a, 0x6b,
447 
+	0x52, 0x16, 0x12, 0x4f, 0x7a, 0x0a, 0xd1, 0x51, 0x27, 0x97, 0x0a, 0x74, 0x69, 0x72, 0x15, 0x77,
448 
+	0x78, 0x36, 0xc7, 0xa6, 0xe3, 0x43, 0x73, 0xb4, 0x91, 0x5a, 0x8a, 0x24, 0x57, 0x0a, 0xf4, 0xe6,
449 
+	0x4a, 0xee, 0x9f, 0x7e, 0x61, 0x56, 0x7a, 0x98, 0xb9, 0x11, 0xfd, 0x1f, 0xc8, 0xdb, 0xee, 0x19,
450 
+	0x2f, 0xfe, 0x79, 0x9e, 0xa4, 0x68, 0x88, 0xe9, 0x1f, 0xa5, 0x49, 0x59, 0x9c, 0x62, 0xe1, 0xda,
451 
+	0xa4, 0x5f, 0x99, 0xc2, 0xb5, 0x93, 0x24, 0x87, 0xd6, 0x79, 0xea, 0x92, 0xfb, 0xca, 0xe4, 0x7a,
452 
+	0xb3, 0xfd, 0x1c, 0x6b, 0xf8, 0x1a, 0x6b, 0xf8, 0x1e, 0x6b, 0x78, 0x59, 0xa7, 0x4b, 0x5d, 0xff,
453 
+	0x04, 0x00, 0x00, 0xff, 0xff, 0x8f, 0xed, 0x9f, 0xb6, 0x38, 0x01, 0x00, 0x00,
449 454 
 }
api/types/plugins/logdriver/entry.proto
History View file @ e0ad6d0
... ... 
@@ -5,4 +5,12 @@ message LogEntry {
5 5 
 	int64 time_nano = 2;
6 6 
 	bytes line = 3;
7 7 
 	bool partial = 4;
8 
+	PartialLogEntryMetadata partial_log_metadata = 5;
8 9 
 }
10 
+
11 
+message PartialLogEntryMetadata {
12 
+	bool last = 1;
13 
+	string id = 2;
14 
+	int32 ordinal = 3;
15 
+}
16 
+
container/container.go
History View file @ e0ad6d0
... ... 
@@ -22,7 +22,9 @@ import (
22 22 
 	"github.com/docker/docker/daemon/exec"
23 23 
 	"github.com/docker/docker/daemon/logger"
24 24 
 	"github.com/docker/docker/daemon/logger/jsonfilelog"
25 
+	"github.com/docker/docker/daemon/logger/local"
25 26 
 	"github.com/docker/docker/daemon/network"
27 
+	"github.com/docker/docker/errdefs"
26 28 
 	"github.com/docker/docker/image"
27 29 
 	"github.com/docker/docker/layer"
28 30 
 	"github.com/docker/docker/pkg/containerfs"
... ... 
@@ -375,13 +377,27 @@ func (container *Container) StartLogger() (logger.Logger, error) {
375 375 
 	}
376 376
377 377 
 	// Set logging file for "json-logger"
378 
-	if cfg.Type == jsonfilelog.Name {
378 
+	// TODO(@cpuguy83): Setup here based on log driver is a little weird.
379 
+	switch cfg.Type {
380 
+	case jsonfilelog.Name:
379 381 
 		info.LogPath, err = container.GetRootResourcePath(fmt.Sprintf("%s-json.log", container.ID))
380 382 
 		if err != nil {
381 383 
 			return nil, err
382 384 
 		}
383 385
384 386 
 		container.LogPath = info.LogPath
387 
+	case local.Name:
388 
+		// Do not set container.LogPath for the local driver
389 
+		// This would expose the value to the API, which should not be done as it means
390 
+		// that the log file implementation would become a stable API that cannot change.
391 
+		logDir, err := container.GetRootResourcePath("local-logs")
392 
+		if err != nil {
393 
+			return nil, err
394 
+		}
395 
+		if err := os.MkdirAll(logDir, 0700); err != nil {
396 
+			return nil, errdefs.System(errors.Wrap(err, "error creating local logs dir"))
397 
+		}
398 
+		info.LogPath = filepath.Join(logDir, "container.log")
385 399 
 	}
386 400
387 401 
 	l, err := initDriver(info)
daemon/logdrivers_linux.go
History View file @ e0ad6d0
... ... 
@@ -9,6 +9,7 @@ import (
9 9 
 	_ "github.com/docker/docker/daemon/logger/gelf"
10 10 
 	_ "github.com/docker/docker/daemon/logger/journald"
11 11 
 	_ "github.com/docker/docker/daemon/logger/jsonfilelog"
12 
+	_ "github.com/docker/docker/daemon/logger/local"
12 13 
 	_ "github.com/docker/docker/daemon/logger/logentries"
13 14 
 	_ "github.com/docker/docker/daemon/logger/splunk"
14 15 
 	_ "github.com/docker/docker/daemon/logger/syslog"
daemon/logger/jsonfilelog/jsonfilelog.go
History View file @ e0ad6d0
... ... 
@@ -110,7 +110,7 @@ func New(info logger.Info) (logger.Logger, error) {
110 110 
 		return b, nil
111 111 
 	}
112 112
113 
-	writer, err := loggerutils.NewLogFile(info.LogPath, capval, maxFiles, compress, marshalFunc, decodeFunc, 0640)
113 
+	writer, err := loggerutils.NewLogFile(info.LogPath, capval, maxFiles, compress, marshalFunc, decodeFunc, 0640, getTailReader)
114 114 
 	if err != nil {
115 115 
 		return nil, err
116 116 
 	}
daemon/logger/jsonfilelog/jsonfilelog_test.go
History View file @ e0ad6d0
... ... 
@@ -4,6 +4,7 @@ import (
4 4 
 	"bytes"
5 5 
 	"compress/gzip"
6 6 
 	"encoding/json"
7 
+	"fmt"
7 8 
 	"io/ioutil"
8 9 
 	"os"
9 10 
 	"path/filepath"
... ... 
@@ -107,7 +108,10 @@ func BenchmarkJSONFileLoggerLog(b *testing.B) {
107 107 
 		ContainerID: "a7317399f3f857173c6179d44823594f8294678dea9999662e5c625b5a1c7657",
108 108 
 		LogPath:     tmp.Join("container.log"),
109 109 
 		Config: map[string]string{
110 
-			"labels": "first,second",
110 
+			"labels":   "first,second",
111 
+			"max-file": "10",
112 
+			"compress": "true",
113 
+			"max-size": "20m",
111 114 
 		},
112 115 
 		ContainerLabels: map[string]string{
113 116 
 			"first":  "label_value",
... ... 
@@ -117,21 +121,34 @@ func BenchmarkJSONFileLoggerLog(b *testing.B) {
117 117 
 	assert.NilError(b, err)
118 118 
 	defer jsonlogger.Close()
119 119
120 
-	msg := &logger.Message{
121 
-		Line:      []byte("Line that thinks that it is log line from docker\n"),
122 
-		Source:    "stderr",
123 
-		Timestamp: time.Now().UTC(),
124 
-	}
125 
-
126 
-	buf := bytes.NewBuffer(nil)
127 
-	assert.NilError(b, marshalMessage(msg, nil, buf))
128 
-	b.SetBytes(int64(buf.Len()))
120 
+	t := time.Now().UTC()
121 
+	for _, data := range [][]byte{
122 
+		[]byte(""),
123 
+		[]byte("a short string"),
124 
+		bytes.Repeat([]byte("a long string"), 100),
125 
+		bytes.Repeat([]byte("a really long string"), 10000),
126 
+	} {
127 
+		b.Run(fmt.Sprintf("%d", len(data)), func(b *testing.B) {
128 
+			testMsg := &logger.Message{
129 
+				Line:      data,
130 
+				Source:    "stderr",
131 
+				Timestamp: t,
132 
+			}
129 133
130 
-	b.ResetTimer()
131 
-	for i := 0; i < b.N; i++ {
132 
-		if err := jsonlogger.Log(msg); err != nil {
133 
-			b.Fatal(err)
134 
-		}
134 
+			buf := bytes.NewBuffer(nil)
135 
+			assert.NilError(b, marshalMessage(testMsg, nil, buf))
136 
+			b.SetBytes(int64(buf.Len()))
137 
+			b.ResetTimer()
138 
+			for i := 0; i < b.N; i++ {
139 
+				msg := logger.NewMessage()
140 
+				msg.Line = testMsg.Line
141 
+				msg.Timestamp = testMsg.Timestamp
142 
+				msg.Source = testMsg.Source
143 
+				if err := jsonlogger.Log(msg); err != nil {
144 
+					b.Fatal(err)
145 
+				}
146 
+			}
147 
+		})
135 148 
 	}
136 149 
 }
137 150
daemon/logger/jsonfilelog/read.go
History View file @ e0ad6d0
... ... 
@@ -1,12 +1,16 @@
1 1 
 package jsonfilelog // import "github.com/docker/docker/daemon/logger/jsonfilelog"
2 2
3 3 
 import (
4 
+	"context"
4 5 
 	"encoding/json"
5 6 
 	"io"
6 7
7 8 
 	"github.com/docker/docker/api/types/backend"
8 9 
 	"github.com/docker/docker/daemon/logger"
9 10 
 	"github.com/docker/docker/daemon/logger/jsonfilelog/jsonlog"
11 
+	"github.com/docker/docker/daemon/logger/loggerutils"
12 
+	"github.com/docker/docker/pkg/tailfile"
13 
+	"github.com/sirupsen/logrus"
10 14 
 )
11 15
12 16 
 const maxJSONDecodeRetry = 20000
... ... 
@@ -63,14 +67,14 @@ func decodeFunc(rdr io.Reader) func() (*logger.Message, error) {
63 63 
 	return func() (msg *logger.Message, err error) {
64 64 
 		for retries := 0; retries < maxJSONDecodeRetry; retries++ {
65 65 
 			msg, err = decodeLogLine(dec, l)
66 
-			if err == nil {
66 
+			if err == nil || err == io.EOF {
67 67 
 				break
68 68 
 			}
69 69
70 
+			logrus.WithError(err).WithField("retries", retries).Warn("got error while decoding json")
70 71 
 			// try again, could be due to a an incomplete json object as we read
71 72 
 			if _, ok := err.(*json.SyntaxError); ok {
72 73 
 				dec = json.NewDecoder(rdr)
73 
-				retries++
74 74 
 				continue
75 75 
 			}
76 76
... ... 
@@ -81,9 +85,13 @@ func decodeFunc(rdr io.Reader) func() (*logger.Message, error) {
81 81 
 			if err == io.ErrUnexpectedEOF {
82 82 
 				reader := io.MultiReader(dec.Buffered(), rdr)
83 83 
 				dec = json.NewDecoder(reader)
84 
-				retries++
84 
+				continue
85 85 
 			}
86 86 
 		}
87 87 
 		return msg, err
88 88 
 	}
89 89 
 }
90 
+
91 
+func getTailReader(ctx context.Context, r loggerutils.SizeReaderAt, req int) (io.Reader, int, error) {
92 
+	return tailfile.NewTailReader(ctx, r, req)
93 
+}
daemon/logger/jsonfilelog/read_test.go
History View file @ e0ad6d0
... ... 
@@ -2,6 +2,7 @@ package jsonfilelog // import "github.com/docker/docker/daemon/logger/jsonfilelo
2 2
3 3 
 import (
4 4 
 	"bytes"
5 
+	"io"
5 6 
 	"testing"
6 7 
 	"time"
7 8
... ... 
@@ -62,3 +63,32 @@ func BenchmarkJSONFileLoggerReadLogs(b *testing.B) {
62 62 
 		}
63 63 
 	}
64 64 
 }
65 
+
66 
+func TestEncodeDecode(t *testing.T) {
67 
+	t.Parallel()
68 
+
69 
+	m1 := &logger.Message{Line: []byte("hello 1"), Timestamp: time.Now(), Source: "stdout"}
70 
+	m2 := &logger.Message{Line: []byte("hello 2"), Timestamp: time.Now(), Source: "stdout"}
71 
+	m3 := &logger.Message{Line: []byte("hello 3"), Timestamp: time.Now(), Source: "stdout"}
72 
+
73 
+	buf := bytes.NewBuffer(nil)
74 
+	assert.Assert(t, marshalMessage(m1, nil, buf))
75 
+	assert.Assert(t, marshalMessage(m2, nil, buf))
76 
+	assert.Assert(t, marshalMessage(m3, nil, buf))
77 
+
78 
+	decode := decodeFunc(buf)
79 
+	msg, err := decode()
80 
+	assert.Assert(t, err)
81 
+	assert.Assert(t, string(msg.Line) == "hello 1\n", string(msg.Line))
82 
+
83 
+	msg, err = decode()
84 
+	assert.Assert(t, err)
85 
+	assert.Assert(t, string(msg.Line) == "hello 2\n")
86 
+
87 
+	msg, err = decode()
88 
+	assert.Assert(t, err)
89 
+	assert.Assert(t, string(msg.Line) == "hello 3\n")
90 
+
91 
+	_, err = decode()
92 
+	assert.Assert(t, err == io.EOF)
93 
+}
daemon/logger/local/config.go
History View file @ e0ad6d0
65 94 
new file mode 100644
... ... 
@@ -0,0 +1,36 @@
0 
+package local
1 
+
2 
+import (
3 
+	"github.com/pkg/errors"
4 
+)
5 
+
6 
+// CreateConfig is used to configure new instances of driver
7 
+type CreateConfig struct {
8 
+	DisableCompression bool
9 
+	MaxFileSize        int64
10 
+	MaxFileCount       int
11 
+}
12 
+
13 
+func newDefaultConfig() *CreateConfig {
14 
+	return &CreateConfig{
15 
+		MaxFileSize:        defaultMaxFileSize,
16 
+		MaxFileCount:       defaultMaxFileCount,
17 
+		DisableCompression: !defaultCompressLogs,
18 
+	}
19 
+}
20 
+
21 
+func validateConfig(cfg *CreateConfig) error {
22 
+	if cfg.MaxFileSize < 0 {
23 
+		return errors.New("max size should be a positive number")
24 
+	}
25 
+	if cfg.MaxFileCount < 0 {
26 
+		return errors.New("max file count cannot be less than 0")
27 
+	}
28 
+
29 
+	if !cfg.DisableCompression {
30 
+		if cfg.MaxFileCount <= 1 {
31 
+			return errors.New("compression cannot be enabled when max file count is 1")
32 
+		}
33 
+	}
34 
+	return nil
35 
+}
daemon/logger/local/doc.go
History View file @ e0ad6d0
0 36 
new file mode 100644
... ... 
@@ -0,0 +1,9 @@
0 
+// Package local provides a logger implementation that stores logs on disk.
1 
+//
2 
+// Log messages are encoded as protobufs with a header and footer for each message.
3 
+// The header and footer are big-endian binary encoded uint32 values which indicate the size of the log message.
4 
+// The header and footer of each message allows you to efficiently read through a file either forwards or in
5 
+// backwards (such as is the case when tailing a file)
6 
+//
7 
+// Example log message format: [22][This is a log message.][22][28][This is another log message.][28]
8 
+package local // import "github.com/docker/docker/daemon/logger/local"
daemon/logger/local/local.go
History View file @ e0ad6d0
0 9 
new file mode 100644
... ... 
@@ -0,0 +1,218 @@
0 
+package local // import "github.com/docker/docker/daemon/logger/local"
1 
+
2 
+import (
3 
+	"encoding/binary"
4 
+	"io"
5 
+	"strconv"
6 
+	"sync"
7 
+	"time"
8 
+
9 
+	"github.com/docker/docker/api/types/backend"
10 
+	"github.com/docker/docker/api/types/plugins/logdriver"
11 
+	"github.com/docker/docker/daemon/logger"
12 
+	"github.com/docker/docker/daemon/logger/loggerutils"
13 
+	"github.com/docker/docker/errdefs"
14 
+	"github.com/docker/go-units"
15 
+	"github.com/pkg/errors"
16 
+	"github.com/sirupsen/logrus"
17 
+)
18 
+
19 
+const (
20 
+	// Name is the name of the driver
21 
+	Name = "local"
22 
+
23 
+	encodeBinaryLen = 4
24 
+	initialBufSize  = 2048
25 
+	maxDecodeRetry  = 20000
26 
+
27 
+	defaultMaxFileSize  int64 = 20 * 1024 * 1024
28 
+	defaultMaxFileCount       = 5
29 
+	defaultCompressLogs       = true
30 
+)
31 
+
32 
+// LogOptKeys are the keys names used for log opts passed in to initialize the driver.
33 
+var LogOptKeys = map[string]bool{
34 
+	"max-file": true,
35 
+	"max-size": true,
36 
+	"compress": true,
37 
+}
38 
+
39 
+// ValidateLogOpt looks for log driver specific options.
40 
+func ValidateLogOpt(cfg map[string]string) error {
41 
+	for key := range cfg {
42 
+		if !LogOptKeys[key] {
43 
+			return errors.Errorf("unknown log opt '%s' for log driver %s", key, Name)
44 
+		}
45 
+	}
46 
+	return nil
47 
+}
48 
+
49 
+func init() {
50 
+	if err := logger.RegisterLogDriver(Name, New); err != nil {
51 
+		logrus.Fatal(err)
52 
+	}
53 
+	if err := logger.RegisterLogOptValidator(Name, ValidateLogOpt); err != nil {
54 
+		logrus.Fatal(err)
55 
+	}
56 
+}
57 
+
58 
+type driver struct {
59 
+	mu      sync.Mutex
60 
+	closed  bool
61 
+	logfile *loggerutils.LogFile
62 
+	readers map[*logger.LogWatcher]struct{} // stores the active log followers
63 
+}
64 
+
65 
+// New creates a new local logger
66 
+// You must provide the `LogPath` in the passed in info argument, this is the file path that logs are written to.
67 
+func New(info logger.Info) (logger.Logger, error) {
68 
+	if info.LogPath == "" {
69 
+		return nil, errdefs.System(errors.New("log path is missing -- this is a bug and should not happen"))
70 
+	}
71 
+
72 
+	cfg := newDefaultConfig()
73 
+	if capacity, ok := info.Config["max-size"]; ok {
74 
+		var err error
75 
+		cfg.MaxFileSize, err = units.FromHumanSize(capacity)
76 
+		if err != nil {
77 
+			return nil, errdefs.InvalidParameter(errors.Wrapf(err, "invalid value for max-size: %s", capacity))
78 
+		}
79 
+	}
80 
+
81 
+	if userMaxFileCount, ok := info.Config["max-file"]; ok {
82 
+		var err error
83 
+		cfg.MaxFileCount, err = strconv.Atoi(userMaxFileCount)
84 
+		if err != nil {
85 
+			return nil, errdefs.InvalidParameter(errors.Wrapf(err, "invalid value for max-file: %s", userMaxFileCount))
86 
+		}
87 
+	}
88 
+
89 
+	if userCompress, ok := info.Config["compress"]; ok {
90 
+		compressLogs, err := strconv.ParseBool(userCompress)
91 
+		if err != nil {
92 
+			return nil, errdefs.InvalidParameter(errors.Wrap(err, "error reading compress log option"))
93 
+		}
94 
+		cfg.DisableCompression = !compressLogs
95 
+	}
96 
+	return newDriver(info.LogPath, cfg)
97 
+}
98 
+
99 
+func makeMarshaller() func(m *logger.Message) ([]byte, error) {
100 
+	buf := make([]byte, initialBufSize)
101 
+
102 
+	// allocate the partial log entry separately, which allows for easier re-use
103 
+	proto := &logdriver.LogEntry{}
104 
+	md := &logdriver.PartialLogEntryMetadata{}
105 
+
106 
+	return func(m *logger.Message) ([]byte, error) {
107 
+		resetProto(proto)
108 
+
109 
+		messageToProto(m, proto, md)
110 
+		protoSize := proto.Size()
111 
+		writeLen := protoSize + (2 * encodeBinaryLen) //+ len(messageDelimiter)
112 
+
113 
+		if writeLen > len(buf) {
114 
+			buf = make([]byte, writeLen)
115 
+		} else {
116 
+			// shrink the buffer back down
117 
+			if writeLen <= initialBufSize {
118 
+				buf = buf[:initialBufSize]
119 
+			} else {
120 
+				buf = buf[:writeLen]
121 
+			}
122 
+		}
123 
+
124 
+		binary.BigEndian.PutUint32(buf[:encodeBinaryLen], uint32(protoSize))
125 
+		n, err := proto.MarshalTo(buf[encodeBinaryLen:writeLen])
126 
+		if err != nil {
127 
+			return nil, errors.Wrap(err, "error marshaling log entry")
128 
+		}
129 
+		if n+(encodeBinaryLen*2) != writeLen {
130 
+			return nil, io.ErrShortWrite
131 
+		}
132 
+		binary.BigEndian.PutUint32(buf[writeLen-encodeBinaryLen:writeLen], uint32(protoSize))
133 
+		return buf[:writeLen], nil
134 
+	}
135 
+}
136 
+
137 
+func newDriver(logPath string, cfg *CreateConfig) (logger.Logger, error) {
138 
+	if err := validateConfig(cfg); err != nil {
139 
+		return nil, errdefs.InvalidParameter(err)
140 
+	}
141 
+
142 
+	lf, err := loggerutils.NewLogFile(logPath, cfg.MaxFileSize, cfg.MaxFileCount, !cfg.DisableCompression, makeMarshaller(), decodeFunc, 0640, getTailReader)
143 
+	if err != nil {
144 
+		return nil, err
145 
+	}
146 
+	return &driver{
147 
+		logfile: lf,
148 
+		readers: make(map[*logger.LogWatcher]struct{}),
149 
+	}, nil
150 
+}
151 
+
152 
+func (d *driver) Name() string {
153 
+	return Name
154 
+}
155 
+
156 
+func (d *driver) Log(msg *logger.Message) error {
157 
+	d.mu.Lock()
158 
+	err := d.logfile.WriteLogEntry(msg)
159 
+	d.mu.Unlock()
160 
+	return err
161 
+}
162 
+
163 
+func (d *driver) Close() error {
164 
+	d.mu.Lock()
165 
+	d.closed = true
166 
+	err := d.logfile.Close()
167 
+	for r := range d.readers {
168 
+		r.Close()
169 
+		delete(d.readers, r)
170 
+	}
171 
+	d.mu.Unlock()
172 
+	return err
173 
+}
174 
+
175 
+func messageToProto(msg *logger.Message, proto *logdriver.LogEntry, partial *logdriver.PartialLogEntryMetadata) {
176 
+	proto.Source = msg.Source
177 
+	proto.TimeNano = msg.Timestamp.UnixNano()
178 
+	proto.Line = append(proto.Line[:0], msg.Line...)
179 
+	proto.Partial = msg.PLogMetaData != nil
180 
+	if proto.Partial {
181 
+		partial.Ordinal = int32(msg.PLogMetaData.Ordinal)
182 
+		partial.Last = msg.PLogMetaData.Last
183 
+		partial.Id = msg.PLogMetaData.ID
184 
+		proto.PartialLogMetadata = partial
185 
+	} else {
186 
+		proto.PartialLogMetadata = nil
187 
+	}
188 
+}
189 
+
190 
+func protoToMessage(proto *logdriver.LogEntry) *logger.Message {
191 
+	msg := &logger.Message{
192 
+		Source:    proto.Source,
193 
+		Timestamp: time.Unix(0, proto.TimeNano),
194 
+	}
195 
+	if proto.Partial {
196 
+		var md backend.PartialLogMetaData
197 
+		md.Last = proto.GetPartialLogMetadata().GetLast()
198 
+		md.ID = proto.GetPartialLogMetadata().GetId()
199 
+		md.Ordinal = int(proto.GetPartialLogMetadata().GetOrdinal())
200 
+		msg.PLogMetaData = &md
201 
+	}
202 
+	msg.Line = append(msg.Line[:0], proto.Line...)
203 
+	return msg
204 
+}
205 
+
206 
+func resetProto(proto *logdriver.LogEntry) {
207 
+	proto.Source = ""
208 
+	proto.Line = proto.Line[:0]
209 
+	proto.TimeNano = 0
210 
+	proto.Partial = false
211 
+	if proto.PartialLogMetadata != nil {
212 
+		proto.PartialLogMetadata.Id = ""
213 
+		proto.PartialLogMetadata.Last = false
214 
+		proto.PartialLogMetadata.Ordinal = 0
215 
+	}
216 
+	proto.PartialLogMetadata = nil
217 
+}
daemon/logger/local/local_test.go
History View file @ e0ad6d0
0 218 
new file mode 100644
... ... 
@@ -0,0 +1,220 @@
0 
+package local
1 
+
2 
+import (
3 
+	"context"
4 
+	"encoding/binary"
5 
+	"io/ioutil"
6 
+	"os"
7 
+	"path/filepath"
8 
+	"testing"
9 
+	"time"
10 
+
11 
+	"bytes"
12 
+	"fmt"
13 
+
14 
+	"strings"
15 
+
16 
+	"io"
17 
+
18 
+	"github.com/docker/docker/api/types/backend"
19 
+	"github.com/docker/docker/api/types/plugins/logdriver"
20 
+	"github.com/docker/docker/daemon/logger"
21 
+	protoio "github.com/gogo/protobuf/io"
22 
+	"gotest.tools/assert"
23 
+	is "gotest.tools/assert/cmp"
24 
+)
25 
+
26 
+func TestWriteLog(t *testing.T) {
27 
+	t.Parallel()
28 
+
29 
+	dir, err := ioutil.TempDir("", t.Name())
30 
+	assert.Assert(t, err)
31 
+	defer os.RemoveAll(dir)
32 
+
33 
+	logPath := filepath.Join(dir, "test.log")
34 
+
35 
+	l, err := New(logger.Info{LogPath: logPath})
36 
+	assert.Assert(t, err)
37 
+	defer l.Close()
38 
+
39 
+	m1 := logger.Message{Source: "stdout", Timestamp: time.Now().Add(-1 * 30 * time.Minute), Line: []byte("message 1")}
40 
+	m2 := logger.Message{Source: "stdout", Timestamp: time.Now().Add(-1 * 20 * time.Minute), Line: []byte("message 2"), PLogMetaData: &backend.PartialLogMetaData{Last: true, ID: "0001", Ordinal: 1}}
41 
+	m3 := logger.Message{Source: "stderr", Timestamp: time.Now().Add(-1 * 10 * time.Minute), Line: []byte("message 3")}
42 
+
43 
+	// copy the log message because the underying log writer resets the log message and returns it to a buffer pool
44 
+	err = l.Log(copyLogMessage(&m1))
45 
+	assert.Assert(t, err)
46 
+	err = l.Log(copyLogMessage(&m2))
47 
+	assert.Assert(t, err)
48 
+	err = l.Log(copyLogMessage(&m3))
49 
+	assert.Assert(t, err)
50 
+
51 
+	f, err := os.Open(logPath)
52 
+	assert.Assert(t, err)
53 
+	defer f.Close()
54 
+	dec := protoio.NewUint32DelimitedReader(f, binary.BigEndian, 1e6)
55 
+
56 
+	var (
57 
+		proto     logdriver.LogEntry
58 
+		testProto logdriver.LogEntry
59 
+		partial   logdriver.PartialLogEntryMetadata
60 
+	)
61 
+
62 
+	lenBuf := make([]byte, encodeBinaryLen)
63 
+	seekMsgLen := func() {
64 
+		io.ReadFull(f, lenBuf)
65 
+	}
66 
+
67 
+	err = dec.ReadMsg(&proto)
68 
+	assert.Assert(t, err)
69 
+	messageToProto(&m1, &testProto, &partial)
70 
+	assert.Check(t, is.DeepEqual(testProto, proto), "expected:\n%+v\ngot:\n%+v", testProto, proto)
71 
+	seekMsgLen()
72 
+
73 
+	err = dec.ReadMsg(&proto)
74 
+	assert.Assert(t, err)
75 
+	messageToProto(&m2, &testProto, &partial)
76 
+	assert.Check(t, is.DeepEqual(testProto, proto))
77 
+	seekMsgLen()
78 
+
79 
+	err = dec.ReadMsg(&proto)
80 
+	assert.Assert(t, err)
81 
+	messageToProto(&m3, &testProto, &partial)
82 
+	assert.Check(t, is.DeepEqual(testProto, proto), "expected:\n%+v\ngot:\n%+v", testProto, proto)
83 
+}
84 
+
85 
+func TestReadLog(t *testing.T) {
86 
+	t.Parallel()
87 
+
88 
+	dir, err := ioutil.TempDir("", t.Name())
89 
+	assert.Assert(t, err)
90 
+	defer os.RemoveAll(dir)
91 
+
92 
+	logPath := filepath.Join(dir, "test.log")
93 
+	l, err := New(logger.Info{LogPath: logPath})
94 
+	assert.Assert(t, err)
95 
+	defer l.Close()
96 
+
97 
+	m1 := logger.Message{Source: "stdout", Timestamp: time.Now().Add(-1 * 30 * time.Minute), Line: []byte("a message")}
98 
+	m2 := logger.Message{Source: "stdout", Timestamp: time.Now().Add(-1 * 20 * time.Minute), Line: []byte("another message"), PLogMetaData: &backend.PartialLogMetaData{Ordinal: 1, Last: true}}
99 
+	longMessage := []byte("a really long message " + strings.Repeat("a", initialBufSize*2))
100 
+	m3 := logger.Message{Source: "stderr", Timestamp: time.Now().Add(-1 * 10 * time.Minute), Line: longMessage}
101 
+	m4 := logger.Message{Source: "stderr", Timestamp: time.Now().Add(-1 * 10 * time.Minute), Line: []byte("just one more message")}
102 
+
103 
+	// copy the log message because the underlying log writer resets the log message and returns it to a buffer pool
104 
+	err = l.Log(copyLogMessage(&m1))
105 
+	assert.Assert(t, err)
106 
+	err = l.Log(copyLogMessage(&m2))
107 
+	assert.Assert(t, err)
108 
+	err = l.Log(copyLogMessage(&m3))
109 
+	assert.Assert(t, err)
110 
+	err = l.Log(copyLogMessage(&m4))
111 
+	assert.Assert(t, err)
112 
+
113 
+	lr := l.(logger.LogReader)
114 
+
115 
+	testMessage := func(t *testing.T, lw *logger.LogWatcher, m *logger.Message) {
116 
+		t.Helper()
117 
+		ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
118 
+		defer cancel()
119 
+		select {
120 
+		case <-ctx.Done():
121 
+			assert.Assert(t, ctx.Err())
122 
+		case err := <-lw.Err:
123 
+			assert.Assert(t, err)
124 
+		case msg, open := <-lw.Msg:
125 
+			if !open {
126 
+				select {
127 
+				case err := <-lw.Err:
128 
+					assert.Assert(t, err)
129 
+				default:
130 
+					assert.Assert(t, m == nil)
131 
+					return
132 
+				}
133 
+			}
134 
+			assert.Assert(t, m != nil)
135 
+			if m.PLogMetaData == nil {
136 
+				// a `\n` is appended on read to make this work with the existing API's when the message is not a partial.
137 
+				// make sure it's the last entry in the line, and then truncate it for the deep equal below.
138 
+				assert.Check(t, msg.Line[len(msg.Line)-1] == '\n')
139 
+				msg.Line = msg.Line[:len(msg.Line)-1]
140 
+			}
141 
+			assert.Check(t, is.DeepEqual(m, msg), fmt.Sprintf("\n%+v\n%+v", m, msg))
142 
+		}
143 
+	}
144 
+
145 
+	t.Run("tail exact", func(t *testing.T) {
146 
+		lw := lr.ReadLogs(logger.ReadConfig{Tail: 4})
147 
+
148 
+		testMessage(t, lw, &m1)
149 
+		testMessage(t, lw, &m2)
150 
+		testMessage(t, lw, &m3)
151 
+		testMessage(t, lw, &m4)
152 
+		testMessage(t, lw, nil) // no more messages
153 
+	})
154 
+
155 
+	t.Run("tail less than available", func(t *testing.T) {
156 
+		lw := lr.ReadLogs(logger.ReadConfig{Tail: 2})
157 
+
158 
+		testMessage(t, lw, &m3)
159 
+		testMessage(t, lw, &m4)
160 
+		testMessage(t, lw, nil) // no more messages
161 
+	})
162 
+
163 
+	t.Run("tail more than available", func(t *testing.T) {
164 
+		lw := lr.ReadLogs(logger.ReadConfig{Tail: 100})
165 
+
166 
+		testMessage(t, lw, &m1)
167 
+		testMessage(t, lw, &m2)
168 
+		testMessage(t, lw, &m3)
169 
+		testMessage(t, lw, &m4)
170 
+		testMessage(t, lw, nil) // no more messages
171 
+	})
172 
+}
173 
+
174 
+func BenchmarkLogWrite(b *testing.B) {
175 
+	f, err := ioutil.TempFile("", b.Name())
176 
+	assert.Assert(b, err)
177 
+	defer os.Remove(f.Name())
178 
+	f.Close()
179 
+
180 
+	local, err := New(logger.Info{LogPath: f.Name()})
181 
+	assert.Assert(b, err)
182 
+	defer local.Close()
183 
+
184 
+	t := time.Now().UTC()
185 
+	for _, data := range [][]byte{
186 
+		[]byte(""),
187 
+		[]byte("a short string"),
188 
+		bytes.Repeat([]byte("a long string"), 100),
189 
+		bytes.Repeat([]byte("a really long string"), 10000),
190 
+	} {
191 
+		b.Run(fmt.Sprintf("%d", len(data)), func(b *testing.B) {
192 
+			entry := &logdriver.LogEntry{Line: data, Source: "stdout", TimeNano: t.UnixNano()}
193 
+			b.SetBytes(int64(entry.Size() + encodeBinaryLen + encodeBinaryLen))
194 
+			b.ResetTimer()
195 
+			for i := 0; i < b.N; i++ {
196 
+				msg := logger.NewMessage()
197 
+				msg.Line = data
198 
+				msg.Timestamp = t
199 
+				msg.Source = "stdout"
200 
+				if err := local.Log(msg); err != nil {
201 
+					b.Fatal(err)
202 
+				}
203 
+			}
204 
+		})
205 
+	}
206 
+}
207 
+
208 
+func copyLogMessage(src *logger.Message) *logger.Message {
209 
+	dst := logger.NewMessage()
210 
+	dst.Source = src.Source
211 
+	dst.Timestamp = src.Timestamp
212 
+	dst.Attrs = src.Attrs
213 
+	dst.Err = src.Err
214 
+	dst.Line = append(dst.Line, src.Line...)
215 
+	if src.PLogMetaData != nil {
216 
+		dst.PLogMetaData = &(*src.PLogMetaData)
217 
+	}
218 
+	return dst
219 
+}
daemon/logger/local/read.go
History View file @ e0ad6d0
0 220 
new file mode 100644
... ... 
@@ -0,0 +1,174 @@
0 
+package local
1 
+
2 
+import (
3 
+	"context"
4 
+	"encoding/binary"
5 
+	"io"
6 
+
7 
+	"bytes"
8 
+
9 
+	"github.com/docker/docker/api/types/plugins/logdriver"
10 
+	"github.com/docker/docker/daemon/logger"
11 
+	"github.com/docker/docker/daemon/logger/loggerutils"
12 
+	"github.com/docker/docker/errdefs"
13 
+	"github.com/pkg/errors"
14 
+)
15 
+
16 
+func (d *driver) ReadLogs(config logger.ReadConfig) *logger.LogWatcher {
17 
+	logWatcher := logger.NewLogWatcher()
18 
+
19 
+	go d.readLogs(logWatcher, config)
20 
+	return logWatcher
21 
+}
22 
+
23 
+func (d *driver) readLogs(watcher *logger.LogWatcher, config logger.ReadConfig) {
24 
+	defer close(watcher.Msg)
25 
+
26 
+	d.mu.Lock()
27 
+	d.readers[watcher] = struct{}{}
28 
+	d.mu.Unlock()
29 
+
30 
+	d.logfile.ReadLogs(config, watcher)
31 
+
32 
+	d.mu.Lock()
33 
+	delete(d.readers, watcher)
34 
+	d.mu.Unlock()
35 
+}
36 
+
37 
+func getTailReader(ctx context.Context, r loggerutils.SizeReaderAt, req int) (io.Reader, int, error) {
38 
+	size := r.Size()
39 
+	if req < 0 {
40 
+		return nil, 0, errdefs.InvalidParameter(errors.Errorf("invalid number of lines to tail: %d", req))
41 
+	}
42 
+
43 
+	if size < (encodeBinaryLen*2)+1 {
44 
+		return bytes.NewReader(nil), 0, nil
45 
+	}
46 
+
47 
+	const encodeBinaryLen64 = int64(encodeBinaryLen)
48 
+	var found int
49 
+
50 
+	buf := make([]byte, encodeBinaryLen)
51 
+
52 
+	offset := size
53 
+	for {
54 
+		select {
55 
+		case <-ctx.Done():
56 
+			return nil, 0, ctx.Err()
57 
+		default:
58 
+		}
59 
+
60 
+		n, err := r.ReadAt(buf, offset-encodeBinaryLen64)
61 
+		if err != nil && err != io.EOF {
62 
+			return nil, 0, errors.Wrap(err, "error reading log message footer")
63 
+		}
64 
+
65 
+		if n != encodeBinaryLen {
66 
+			return nil, 0, errdefs.DataLoss(errors.New("unexpected number of bytes read from log message footer"))
67 
+		}
68 
+
69 
+		msgLen := binary.BigEndian.Uint32(buf)
70 
+
71 
+		n, err = r.ReadAt(buf, offset-encodeBinaryLen64-encodeBinaryLen64-int64(msgLen))
72 
+		if err != nil && err != io.EOF {
73 
+			return nil, 0, errors.Wrap(err, "error reading log message header")
74 
+		}
75 
+
76 
+		if n != encodeBinaryLen {
77 
+			return nil, 0, errdefs.DataLoss(errors.New("unexpected number of bytes read from log message header"))
78 
+		}
79 
+
80 
+		if msgLen != binary.BigEndian.Uint32(buf) {
81 
+			return nil, 0, errdefs.DataLoss(errors.Wrap(err, "log message header and footer indicate different message sizes"))
82 
+		}
83 
+
84 
+		found++
85 
+		offset -= int64(msgLen)
86 
+		offset -= encodeBinaryLen64 * 2
87 
+		if found == req {
88 
+			break
89 
+		}
90 
+		if offset <= 0 {
91 
+			break
92 
+		}
93 
+	}
94 
+
95 
+	return io.NewSectionReader(r, offset, size), found, nil
96 
+}
97 
+
98 
+func decodeFunc(rdr io.Reader) func() (*logger.Message, error) {
99 
+	proto := &logdriver.LogEntry{}
100 
+	buf := make([]byte, initialBufSize)
101 
+
102 
+	return func() (*logger.Message, error) {
103 
+		var (
104 
+			read int
105 
+			err  error
106 
+		)
107 
+
108 
+		resetProto(proto)
109 
+
110 
+		for i := 0; i < maxDecodeRetry; i++ {
111 
+			var n int
112 
+			n, err = io.ReadFull(rdr, buf[read:encodeBinaryLen])
113 
+			if err != nil {
114 
+				if err != io.ErrUnexpectedEOF {
115 
+					return nil, errors.Wrap(err, "error reading log message length")
116 
+				}
117 
+				read += n
118 
+				continue
119 
+			}
120 
+			read += n
121 
+			break
122 
+		}
123 
+		if err != nil {
124 
+			return nil, errors.Wrapf(err, "could not read log message length: read: %d, expected: %d", read, encodeBinaryLen)
125 
+		}
126 
+
127 
+		msgLen := int(binary.BigEndian.Uint32(buf[:read]))
128 
+
129 
+		if len(buf) < msgLen+encodeBinaryLen {
130 
+			buf = make([]byte, msgLen+encodeBinaryLen)
131 
+		} else {
132 
+			if msgLen <= initialBufSize {
133 
+				buf = buf[:initialBufSize]
134 
+			} else {
135 
+				buf = buf[:msgLen+encodeBinaryLen]
136 
+			}
137 
+		}
138 
+
139 
+		return decodeLogEntry(rdr, proto, buf, msgLen)
140 
+	}
141 
+}
142 
+
143 
+func decodeLogEntry(rdr io.Reader, proto *logdriver.LogEntry, buf []byte, msgLen int) (*logger.Message, error) {
144 
+	var (
145 
+		read int
146 
+		err  error
147 
+	)
148 
+	for i := 0; i < maxDecodeRetry; i++ {
149 
+		var n int
150 
+		n, err = io.ReadFull(rdr, buf[read:msgLen+encodeBinaryLen])
151 
+		if err != nil {
152 
+			if err != io.ErrUnexpectedEOF {
153 
+				return nil, errors.Wrap(err, "could not decode log entry")
154 
+			}
155 
+			read += n
156 
+			continue
157 
+		}
158 
+		break
159 
+	}
160 
+	if err != nil {
161 
+		return nil, errors.Wrapf(err, "could not decode entry: read %d, expected: %d", read, msgLen)
162 
+	}
163 
+
164 
+	if err := proto.Unmarshal(buf[:msgLen]); err != nil {
165 
+		return nil, errors.Wrap(err, "error unmarshalling log entry")
166 
+	}
167 
+
168 
+	msg := protoToMessage(proto)
169 
+	if msg.PLogMetaData == nil {
170 
+		msg.Line = append(msg.Line, '\n')
171 
+	}
172 
+	return msg, nil
173 
+}
daemon/logger/loggerutils/logfile.go
History View file @ e0ad6d0
... ... 
@@ -1,7 +1,6 @@
1 1 
 package loggerutils // import "github.com/docker/docker/daemon/logger/loggerutils"
2 2
3 3 
 import (
4 
-	"bytes"
5 4 
 	"compress/gzip"
6 5 
 	"context"
7 6 
 	"encoding/json"
... ... 
@@ -15,11 +14,9 @@ import (
15 15 
 	"time"
16 16
17 17 
 	"github.com/docker/docker/daemon/logger"
18 
-	"github.com/docker/docker/daemon/logger/loggerutils/multireader"
19 18 
 	"github.com/docker/docker/pkg/filenotify"
20 19 
 	"github.com/docker/docker/pkg/pools"
21 20 
 	"github.com/docker/docker/pkg/pubsub"
22 
-	"github.com/docker/docker/pkg/tailfile"
23 21 
 	"github.com/fsnotify/fsnotify"
24 22 
 	"github.com/pkg/errors"
25 23 
 	"github.com/sirupsen/logrus"
... ... 
@@ -93,13 +90,27 @@ type LogFile struct {
93 93 
 	notifyRotate    *pubsub.Publisher
94 94 
 	marshal         logger.MarshalFunc
95 95 
 	createDecoder   makeDecoderFunc
96 
+	getTailReader   GetTailReaderFunc
96 97 
 	perms           os.FileMode
97 98 
 }
98 99
99 100 
 type makeDecoderFunc func(rdr io.Reader) func() (*logger.Message, error)
100 101
102 
+// SizeReaderAt defines a ReaderAt that also reports its size.
103 
+// This is used for tailing log files.
104 
+type SizeReaderAt interface {
105 
+	io.ReaderAt
106 
+	Size() int64
107 
+}
108 
+
109 
+// GetTailReaderFunc is used to truncate a reader to only read as much as is required
110 
+// in order to get the passed in number of log lines.
111 
+// It returns the sectioned reader, the number of lines that the section reader
112 
+// contains, and any error that occurs.
113 
+type GetTailReaderFunc func(ctx context.Context, f SizeReaderAt, nLogLines int) (rdr io.Reader, nLines int, err error)
114 
+
101 115 
 // NewLogFile creates new LogFile
102 
-func NewLogFile(logPath string, capacity int64, maxFiles int, compress bool, marshaller logger.MarshalFunc, decodeFunc makeDecoderFunc, perms os.FileMode) (*LogFile, error) {
116 
+func NewLogFile(logPath string, capacity int64, maxFiles int, compress bool, marshaller logger.MarshalFunc, decodeFunc makeDecoderFunc, perms os.FileMode, getTailReader GetTailReaderFunc) (*LogFile, error) {
103 117 
 	log, err := os.OpenFile(logPath, os.O_WRONLY|os.O_APPEND|os.O_CREATE, perms)
104 118 
 	if err != nil {
105 119 
 		return nil, err
... ... 
@@ -121,6 +132,7 @@ func NewLogFile(logPath string, capacity int64, maxFiles int, compress bool, mar
121 121 
 		marshal:         marshaller,
122 122 
 		createDecoder:   decodeFunc,
123 123 
 		perms:           perms,
124 
+		getTailReader:   getTailReader,
124 125 
 	}, nil
125 126 
 }
126 127
... ... 
@@ -310,33 +322,45 @@ func (w *LogFile) ReadLogs(config logger.ReadConfig, watcher *logger.LogWatcher)
310 310 
 	}
311 311
312 312 
 	if config.Tail != 0 {
313 
+		// TODO(@cpuguy83): Instead of opening every file, only get the files which
314 
+		// are needed to tail.
315 
+		// This is especially costly when compression is enabled.
313 316 
 		files, err := w.openRotatedFiles(config)
317 
+		w.mu.RUnlock()
314 318 
 		if err != nil {
315 
-			w.mu.RUnlock()
316 319 
 			watcher.Err <- err
317 320 
 			return
318 321 
 		}
319 
-		w.mu.RUnlock()
320 
-		seekers := make([]io.ReadSeeker, 0, len(files)+1)
321 
-		for _, f := range files {
322 
-			seekers = append(seekers, f)
323 
-		}
324 
-		if currentChunk.Size() > 0 {
325 
-			seekers = append(seekers, currentChunk)
326 
-		}
327 
-		if len(seekers) > 0 {
328 
-			tailFile(multireader.MultiReadSeeker(seekers...), watcher, w.createDecoder, config)
322 
+
323 
+		closeFiles := func() {
324 
+			for _, f := range files {
325 
+				f.Close()
326 
+				fileName := f.Name()
327 
+				if strings.HasSuffix(fileName, tmpLogfileSuffix) {
328 
+					err := w.filesRefCounter.Dereference(fileName)
329 
+					if err != nil {
330 
+						logrus.Errorf("Failed to dereference the log file %q: %v", fileName, err)
331 
+					}
332 
+				}
333 
+			}
329 334 
 		}
335 
+
336 
+		readers := make([]SizeReaderAt, 0, len(files)+1)
330 337 
 		for _, f := range files {
331 
-			f.Close()
332 
-			fileName := f.Name()
333 
-			if strings.HasSuffix(fileName, tmpLogfileSuffix) {
334 
-				err := w.filesRefCounter.Dereference(fileName)
335 
-				if err != nil {
336 
-					logrus.Errorf("Failed to dereference log file %q: %v", fileName, err)
337 
-				}
338 
+			stat, err := f.Stat()
339 
+			if err != nil {
340 
+				watcher.Err <- errors.Wrap(err, "error reading size of rotated file")
341 
+				closeFiles()
342 
+				return
338 343 
 			}
344 
+			readers = append(readers, io.NewSectionReader(f, 0, stat.Size()))
339 345 
 		}
346 
+		if currentChunk.Size() > 0 {
347 
+			readers = append(readers, currentChunk)
348 
+		}
349 
+
350 
+		tailFiles(readers, watcher, w.createDecoder, w.getTailReader, config)
351 
+		closeFiles()
340 352
341 353 
 		w.mu.RLock()
342 354 
 	}
... ... 
@@ -455,19 +479,39 @@ func newSectionReader(f *os.File) (*io.SectionReader, error) {
455 455 
 	return io.NewSectionReader(f, 0, size), nil
456 456 
 }
457 457
458 
-type decodeFunc func() (*logger.Message, error)
458 
+func tailFiles(files []SizeReaderAt, watcher *logger.LogWatcher, createDecoder makeDecoderFunc, getTailReader GetTailReaderFunc, config logger.ReadConfig) {
459 
+	nLines := config.Tail
460 
+
461 
+	ctx, cancel := context.WithCancel(context.Background())
462 
+	defer cancel()
463 
+	// TODO(@cpuguy83): we should plumb a context through instead of dealing with `WatchClose()` here.
464 
+	go func() {
465 
+		select {
466 
+		case <-ctx.Done():
467 
+		case <-watcher.WatchClose():
468 
+			cancel()
469 
+		}
470 
+	}()
471 
+
472 
+	readers := make([]io.Reader, 0, len(files))
459 473
460 
-func tailFile(f io.ReadSeeker, watcher *logger.LogWatcher, createDecoder makeDecoderFunc, config logger.ReadConfig) {
461 
-	var rdr io.Reader = f
462 474 
 	if config.Tail > 0 {
463 
-		ls, err := tailfile.TailFile(f, config.Tail)
464 
-		if err != nil {
465 
-			watcher.Err <- err
466 
-			return
475 
+		for i := len(files) - 1; i >= 0 && nLines > 0; i-- {
476 
+			tail, n, err := getTailReader(ctx, files[i], nLines)
477 
+			if err != nil {
478 
+				watcher.Err <- errors.Wrap(err, "error finding file position to start log tailing")
479 
+				return
480 
+			}
481 
+			nLines -= n
482 
+			readers = append([]io.Reader{tail}, readers...)
483 
+		}
484 
+	} else {
485 
+		for _, r := range files {
486 
+			readers = append(readers, &wrappedReaderAt{ReaderAt: r})
467 487 
 		}
468 
-		rdr = bytes.NewBuffer(bytes.Join(ls, []byte("\n")))
469 488 
 	}
470 489
490 
+	rdr := io.MultiReader(readers...)
471 491 
 	decodeLogLine := createDecoder(rdr)
472 492 
 	for {
473 493 
 		msg, err := decodeLogLine()
... ... 
@@ -484,7 +528,7 @@ func tailFile(f io.ReadSeeker, watcher *logger.LogWatcher, createDecoder makeDec
484 484 
 			return
485 485 
 		}
486 486 
 		select {
487 
-		case <-watcher.WatchClose():
487 
+		case <-ctx.Done():
488 488 
 			return
489 489 
 		case watcher.Msg <- msg:
490 490 
 		}
... ... 
@@ -678,3 +722,14 @@ func watchFile(name string) (filenotify.FileWatcher, error) {
678 678
679 679 
 	return fileWatcher, nil
680 680 
 }
681 
+
682 
+type wrappedReaderAt struct {
683 
+	io.ReaderAt
684 
+	pos int64
685 
+}
686 
+
687 
+func (r *wrappedReaderAt) Read(p []byte) (int, error) {
688 
+	n, err := r.ReaderAt.ReadAt(p, r.pos)
689 
+	r.pos += int64(n)
690 
+	return n, err
691 
+}
daemon/logger/loggerutils/logfile_test.go
History View file @ e0ad6d0
681 692 
new file mode 100644
... ... 
@@ -0,0 +1,76 @@
0 
+package loggerutils
1 
+
2 
+import (
3 
+	"bufio"
4 
+	"context"
5 
+	"io"
6 
+	"strings"
7 
+	"testing"
8 
+	"time"
9 
+
10 
+	"github.com/docker/docker/daemon/logger"
11 
+	"github.com/docker/docker/pkg/tailfile"
12 
+	"gotest.tools/assert"
13 
+)
14 
+
15 
+func TestTailFiles(t *testing.T) {
16 
+	s1 := strings.NewReader("Hello.\nMy name is Inigo Montoya.\n")
17 
+	s2 := strings.NewReader("I'm serious.\nDon't call me Shirley!\n")
18 
+	s3 := strings.NewReader("Roads?\nWhere we're going we don't need roads.\n")
19 
+
20 
+	files := []SizeReaderAt{s1, s2, s3}
21 
+	watcher := logger.NewLogWatcher()
22 
+	createDecoder := func(r io.Reader) func() (*logger.Message, error) {
23 
+		scanner := bufio.NewScanner(r)
24 
+		return func() (*logger.Message, error) {
25 
+			if !scanner.Scan() {
26 
+				return nil, scanner.Err()
27 
+			}
28 
+			// some comment
29 
+			return &logger.Message{Line: scanner.Bytes(), Timestamp: time.Now()}, nil
30 
+		}
31 
+	}
32 
+
33 
+	tailReader := func(ctx context.Context, r SizeReaderAt, lines int) (io.Reader, int, error) {
34 
+		return tailfile.NewTailReader(ctx, r, lines)
35 
+	}
36 
+
37 
+	for desc, config := range map[string]logger.ReadConfig{} {
38 
+		t.Run(desc, func(t *testing.T) {
39 
+			started := make(chan struct{})
40 
+			go func() {
41 
+				close(started)
42 
+				tailFiles(files, watcher, createDecoder, tailReader, config)
43 
+			}()
44 
+			<-started
45 
+		})
46 
+	}
47 
+
48 
+	config := logger.ReadConfig{Tail: 2}
49 
+	started := make(chan struct{})
50 
+	go func() {
51 
+		close(started)
52 
+		tailFiles(files, watcher, createDecoder, tailReader, config)
53 
+	}()
54 
+	<-started
55 
+
56 
+	select {
57 
+	case <-time.After(60 * time.Second):
58 
+		t.Fatal("timeout waiting for tail line")
59 
+	case err := <-watcher.Err:
60 
+		assert.Assert(t, err)
61 
+	case msg := <-watcher.Msg:
62 
+		assert.Assert(t, msg != nil)
63 
+		assert.Assert(t, string(msg.Line) == "Roads?", string(msg.Line))
64 
+	}
65 
+
66 
+	select {
67 
+	case <-time.After(60 * time.Second):
68 
+		t.Fatal("timeout waiting for tail line")
69 
+	case err := <-watcher.Err:
70 
+		assert.Assert(t, err)
71 
+	case msg := <-watcher.Msg:
72 
+		assert.Assert(t, msg != nil)
73 
+		assert.Assert(t, string(msg.Line) == "Where we're going we don't need roads.", string(msg.Line))
74 
+	}
75 
+}
pkg/tailfile/tailfile.go
History View file @ e0ad6d0
... ... 
@@ -3,7 +3,9 @@
3 3 
 package tailfile // import "github.com/docker/docker/pkg/tailfile"
4 4
5 5 
 import (
6 
+	"bufio"
6 7 
 	"bytes"
8 
+	"context"
7 9 
 	"errors"
8 10 
 	"io"
9 11 
 	"os"
... ... 
@@ -16,51 +18,205 @@ var eol = []byte("\n")
16 16 
 // ErrNonPositiveLinesNumber is an error returned if the lines number was negative.
17 17 
 var ErrNonPositiveLinesNumber = errors.New("The number of lines to extract from the file must be positive")
18 18
19 
-//TailFile returns last n lines of reader f (could be a nil).
20 
-func TailFile(f io.ReadSeeker, n int) ([][]byte, error) {
21 
-	if n <= 0 {
22 
-		return nil, ErrNonPositiveLinesNumber
19 
+//TailFile returns last n lines of the passed in file.
20 
+func TailFile(f *os.File, n int) ([][]byte, error) {
21 
+	size, err := f.Seek(0, io.SeekEnd)
22 
+	if err != nil {
23 
+		return nil, err
23 24 
 	}
24 
-	size, err := f.Seek(0, os.SEEK_END)
25 
+
26 
+	rAt := io.NewSectionReader(f, 0, size)
27 
+	r, nLines, err := NewTailReader(context.Background(), rAt, n)
25 28 
 	if err != nil {
26 29 
 		return nil, err
27 30 
 	}
28 
-	block := -1
29 
-	var data []byte
30 
-	var cnt int
31 
+
32 
+	buf := make([][]byte, 0, nLines)
33 
+	scanner := bufio.NewScanner(r)
34 
+
35 
+	for scanner.Scan() {
36 
+		buf = append(buf, scanner.Bytes())
37 
+	}
38 
+	return buf, nil
39 
+}
40 
+
41 
+// SizeReaderAt is an interface used to get a ReaderAt as well as the size of the underlying reader.
42 
+// Note that the size of the underlying reader should not change when using this interface.
43 
+type SizeReaderAt interface {
44 
+	io.ReaderAt
45 
+	Size() int64
46 
+}
47 
+
48 
+// NewTailReader scopes the passed in reader to just the last N lines passed in
49 
+func NewTailReader(ctx context.Context, r SizeReaderAt, reqLines int) (io.Reader, int, error) {
50 
+	return NewTailReaderWithDelimiter(ctx, r, reqLines, eol)
51 
+}
52 
+
53 
+// NewTailReaderWithDelimiter scopes the passed in reader to just the last N lines passed in
54 
+// In this case a "line" is defined by the passed in delimiter.
55 
+//
56 
+// Delimiter lengths should be generally small, no more than 12 bytes
57 
+func NewTailReaderWithDelimiter(ctx context.Context, r SizeReaderAt, reqLines int, delimiter []byte) (io.Reader, int, error) {
58 
+	if reqLines < 1 {
59 
+		return nil, 0, ErrNonPositiveLinesNumber
60 
+	}
61 
+	if len(delimiter) == 0 {
62 
+		return nil, 0, errors.New("must provide a delimiter")
63 
+	}
64 
+	var (
65 
+		size      = r.Size()
66 
+		tailStart int64
67 
+		tailEnd   = size
68 
+		found     int
69 
+	)
70 
+
71 
+	if int64(len(delimiter)) >= size {
72 
+		return bytes.NewReader(nil), 0, nil
73 
+	}
74 
+
75 
+	scanner := newScanner(r, delimiter)
76 
+	for scanner.Scan(ctx) {
77 
+		if err := scanner.Err(); err != nil {
78 
+			return nil, 0, scanner.Err()
79 
+		}
80 
+
81 
+		found++
82 
+		if found == 1 {
83 
+			tailEnd = scanner.End()
84 
+		}
85 
+		if found == reqLines {
86 
+			break
87 
+		}
88 
+	}
89 
+
90 
+	tailStart = scanner.Start(ctx)
91 
+
92 
+	if found == 0 {
93 
+		return bytes.NewReader(nil), 0, nil
94 
+	}
95 
+
96 
+	if found < reqLines && tailStart != 0 {
97 
+		tailStart = 0
98 
+	}
99 
+	return io.NewSectionReader(r, tailStart, tailEnd-tailStart), found, nil
100 
+}
101 
+
102 
+func newScanner(r SizeReaderAt, delim []byte) *scanner {
103 
+	size := r.Size()
104 
+	readSize := blockSize
105 
+	if readSize > int(size) {
106 
+		readSize = int(size)
107 
+	}
108 
+	// silly case...
109 
+	if len(delim) >= readSize/2 {
110 
+		readSize = len(delim)*2 + 2
111 
+	}
112 
+
113 
+	return &scanner{
114 
+		r:     r,
115 
+		pos:   size,
116 
+		buf:   make([]byte, readSize),
117 
+		delim: delim,
118 
+	}
119 
+}
120 
+
121 
+type scanner struct {
122 
+	r     SizeReaderAt
123 
+	pos   int64
124 
+	buf   []byte
125 
+	delim []byte
126 
+	err   error
127 
+	idx   int
128 
+	done  bool
129 
+}
130 
+
131 
+func (s *scanner) Start(ctx context.Context) int64 {
132 
+	if s.idx > 0 {
133 
+		idx := bytes.LastIndex(s.buf[:s.idx], s.delim)
134 
+		if idx >= 0 {
135 
+			return s.pos + int64(idx) + int64(len(s.delim))
136 
+		}
137 
+	}
138 
+
139 
+	// slow path
140 
+	buf := make([]byte, len(s.buf))
141 
+	copy(buf, s.buf)
142 
+
143 
+	readAhead := &scanner{
144 
+		r:     s.r,
145 
+		pos:   s.pos,
146 
+		delim: s.delim,
147 
+		idx:   s.idx,
148 
+		buf:   buf,
149 
+	}
150 
+
151 
+	if !readAhead.Scan(ctx) {
152 
+		return 0
153 
+	}
154 
+	return readAhead.End()
155 
+}
156 
+
157 
+func (s *scanner) End() int64 {
158 
+	return s.pos + int64(s.idx) + int64(len(s.delim))
159 
+}
160 
+
161 
+func (s *scanner) Err() error {
162 
+	return s.err
163 
+}
164 
+
165 
+func (s *scanner) Scan(ctx context.Context) bool {
166 
+	if s.err != nil {
167 
+		return false
168 
+	}
169 
+
31 170 
 	for {
32 
-		var b []byte
33 
-		step := int64(block * blockSize)
34 
-		left := size + step // how many bytes to beginning
35 
-		if left < 0 {
36 
-			if _, err := f.Seek(0, os.SEEK_SET); err != nil {
37 
-				return nil, err
38 
-			}
39 
-			b = make([]byte, blockSize+left)
40 
-			if _, err := f.Read(b); err != nil {
41 
-				return nil, err
171 
+		select {
172 
+		case <-ctx.Done():
173 
+			s.err = ctx.Err()
174 
+			return false
175 
+		default:
176 
+		}
177 
+
178 
+		idx := s.idx - len(s.delim)
179 
+		if idx < 0 {
180 
+			readSize := int(s.pos)
181 
+			if readSize > len(s.buf) {
182 
+				readSize = len(s.buf)
42 183 
 			}
43 
-			data = append(b, data...)
44 
-			break
45 
-		} else {
46 
-			b = make([]byte, blockSize)
47 
-			if _, err := f.Seek(left, os.SEEK_SET); err != nil {
48 
-				return nil, err
184 
+
185 
+			if readSize < len(s.delim) {
186 
+				return false
49 187 
 			}
50 
-			if _, err := f.Read(b); err != nil {
51 
-				return nil, err
188 
+
189 
+			offset := s.pos - int64(readSize)
190 
+			n, err := s.r.ReadAt(s.buf[:readSize], offset)
191 
+			if err != nil && err != io.EOF {
192 
+				s.err = err
193 
+				return false
52 194 
 			}
53 
-			data = append(b, data...)
195 
+
196 
+			s.pos -= int64(n)
197 
+			idx = n
54 198 
 		}
55 
-		cnt += bytes.Count(b, eol)
56 
-		if cnt > n {
57 
-			break
199 
+
200 
+		s.idx = bytes.LastIndex(s.buf[:idx], s.delim)
201 
+		if s.idx >= 0 {
202 
+			return true
58 203 
 		}
59 
-		block--
60 
-	}
61 
-	lines := bytes.Split(data, eol)
62 
-	if n < len(lines) {
63 
-		return lines[len(lines)-n-1 : len(lines)-1], nil
204 
+
205 
+		if len(s.delim) > 1 && s.pos > 0 {
206 
+			// in this case, there may be a partial delimiter at the front of the buffer, so set the position forward
207 
+			// up to the maximum size partial that could be there so it can be read again in the next iteration with any
208 
+			// potential remainder.
209 
+			// An example where delimiter is `####`:
210 
+			// [##asdfqwerty]
211 
+			//    ^
212 
+			// This resets the position to where the arrow is pointing.
213 
+			// It could actually check if a partial exists and at the front, but that is pretty similar to the indexing
214 
+			// code above though a bit more complex since each byte has to be checked (`len(delimiter)-1`) factorial).
215 
+			// It's much simpler and cleaner to just re-read `len(delimiter)-1` bytes again.
216 
+			s.pos += int64(len(s.delim)) - 1
217 
+		}
218 
+
64 219 
 	}
65 
-	return lines[:len(lines)-1], nil
66 220 
 }
pkg/tailfile/tailfile_test.go
History View file @ e0ad6d0
... ... 
@@ -1,9 +1,17 @@
1 1 
 package tailfile // import "github.com/docker/docker/pkg/tailfile"
2 2
3 3 
 import (
4 
+	"bufio"
5 
+	"bytes"
6 
+	"context"
7 
+	"fmt"
8 
+	"io"
4 9 
 	"io/ioutil"
5 10 
 	"os"
11 
+	"strings"
6 12 
 	"testing"
13 
+
14 
+	"gotest.tools/assert"
7 15 
 )
8 16
9 17 
 func TestTailFile(t *testing.T) {
... ... 
@@ -42,7 +50,7 @@ truncated line`)
42 42 
 	if _, err := f.Write(testFile); err != nil {
43 43 
 		t.Fatal(err)
44 44 
 	}
45 
-	if _, err := f.Seek(0, os.SEEK_SET); err != nil {
45 
+	if _, err := f.Seek(0, io.SeekStart); err != nil {
46 46 
 		t.Fatal(err)
47 47 
 	}
48 48 
 	expected := []string{"last fourth line", "last fifth line"}
... ... 
@@ -50,10 +58,12 @@ truncated line`)
50 50 
 	if err != nil {
51 51 
 		t.Fatal(err)
52 52 
 	}
53 
+	if len(res) != len(expected) {
54 
+		t.Fatalf("\nexpected:\n%s\n\nactual:\n%s", expected, res)
55 
+	}
53 56 
 	for i, l := range res {
54 
-		t.Logf("%s", l)
55 57 
 		if expected[i] != string(l) {
56 
-			t.Fatalf("Expected line %s, got %s", expected[i], l)
58 
+			t.Fatalf("Expected line %q, got %q", expected[i], l)
57 59 
 		}
58 60 
 	}
59 61 
 }
... ... 
@@ -71,7 +81,7 @@ truncated line`)
71 71 
 	if _, err := f.Write(testFile); err != nil {
72 72 
 		t.Fatal(err)
73 73 
 	}
74 
-	if _, err := f.Seek(0, os.SEEK_SET); err != nil {
74 
+	if _, err := f.Seek(0, io.SeekStart); err != nil {
75 75 
 		t.Fatal(err)
76 76 
 	}
77 77 
 	expected := []string{"first line", "second line"}
... ... 
@@ -79,8 +89,10 @@ truncated line`)
79 79 
 	if err != nil {
80 80 
 		t.Fatal(err)
81 81 
 	}
82 
+	if len(expected) != len(res) {
83 
+		t.Fatalf("\nexpected:\n%s\n\nactual:\n%s", expected, res)
84 
+	}
82 85 
 	for i, l := range res {
83 
-		t.Logf("%s", l)
84 86 
 		if expected[i] != string(l) {
85 87 
 			t.Fatalf("Expected line %s, got %s", expected[i], l)
86 88 
 		}
... ... 
@@ -116,11 +128,11 @@ truncated line`)
116 116 
 	if _, err := f.Write(testFile); err != nil {
117 117 
 		t.Fatal(err)
118 118 
 	}
119 
-	if _, err := f.Seek(0, os.SEEK_SET); err != nil {
119 
+	if _, err := f.Seek(0, io.SeekStart); err != nil {
120 120 
 		t.Fatal(err)
121 121 
 	}
122 122 
 	if _, err := TailFile(f, -1); err != ErrNonPositiveLinesNumber {
123 
-		t.Fatalf("Expected ErrNonPositiveLinesNumber, got %s", err)
123 
+		t.Fatalf("Expected ErrNonPositiveLinesNumber, got %v", err)
124 124 
 	}
125 125 
 	if _, err := TailFile(f, 0); err != ErrNonPositiveLinesNumber {
126 126 
 		t.Fatalf("Expected ErrNonPositiveLinesNumber, got %s", err)
... ... 
@@ -146,3 +158,170 @@ func BenchmarkTail(b *testing.B) {
146 146 
 		}
147 147 
 	}
148 148 
 }
149 
+
150 
+func TestNewTailReader(t *testing.T) {
151 
+	t.Parallel()
152 
+	ctx := context.Background()
153 
+
154 
+	for dName, delim := range map[string][]byte{
155 
+		"no delimiter":          {},
156 
+		"single byte delimiter": {'\n'},
157 
+		"2 byte delimiter":      []byte(";\n"),
158 
+		"4 byte delimiter":      []byte("####"),
159 
+		"8 byte delimiter":      []byte("########"),
160 
+		"12 byte delimiter":     []byte("############"),
161 
+	} {
162 
+		t.Run(dName, func(t *testing.T) {
163 
+			delim := delim
164 
+			t.Parallel()
165 
+
166 
+			s1 := "Hello world."
167 
+			s2 := "Today is a fine day."
168 
+			s3 := "So long, and thanks for all the fish!"
169 
+			s4 := strings.Repeat("a", blockSize/2) // same as block size
170 
+			s5 := strings.Repeat("a", blockSize)   // just to make sure
171 
+			s6 := strings.Repeat("a", blockSize*2) // bigger than block size
172 
+			s7 := strings.Repeat("a", blockSize-1) // single line same as block
173 
+
174 
+			s8 := `{"log":"Don't panic!\n","stream":"stdout","time":"2018-04-04T20:28:44.7207062Z"}`
175 
+			jsonTest := make([]string, 0, 20)
176 
+			for i := 0; i < 20; i++ {
177 
+				jsonTest = append(jsonTest, s8)
178 
+			}
179 
+
180 
+			for _, test := range []struct {
181 
+				desc string
182 
+				data []string
183 
+			}{
184 
+				{desc: "one small entry", data: []string{s1}},
185 
+				{desc: "several small entries", data: []string{s1, s2, s3}},
186 
+				{desc: "various sizes", data: []string{s1, s2, s3, s4, s5, s1, s2, s3, s7, s6}},
187 
+				{desc: "multiple lines with one more than block", data: []string{s5, s5, s5, s5, s5}},
188 
+				{desc: "multiple lines much bigger than block", data: []string{s6, s6, s6, s6, s6}},
189 
+				{desc: "multiple lines same as block", data: []string{s4, s4, s4, s4, s4}},
190 
+				{desc: "single line same as block", data: []string{s7}},
191 
+				{desc: "single line half block", data: []string{s4}},
192 
+				{desc: "single line twice block", data: []string{s6}},
193 
+				{desc: "json encoded values", data: jsonTest},
194 
+				{desc: "no lines", data: []string{}},
195 
+				{desc: "same length as delimiter", data: []string{strings.Repeat("a", len(delim))}},
196 
+			} {
197 
+				t.Run(test.desc, func(t *testing.T) {
198 
+					test := test
199 
+					t.Parallel()
200 
+
201 
+					max := len(test.data)
202 
+					if max > 10 {
203 
+						max = 10
204 
+					}
205 
+
206 
+					s := strings.Join(test.data, string(delim))
207 
+					if len(test.data) > 0 {
208 
+						s += string(delim)
209 
+					}
210 
+
211 
+					for i := 1; i <= max; i++ {
212 
+						t.Run(fmt.Sprintf("%d lines", i), func(t *testing.T) {
213 
+							i := i
214 
+							t.Parallel()
215 
+
216 
+							r := strings.NewReader(s)
217 
+							tr, lines, err := NewTailReaderWithDelimiter(ctx, r, i, delim)
218 
+							if len(delim) == 0 {
219 
+								assert.Assert(t, err != nil)
220 
+								assert.Assert(t, lines == 0)
221 
+								return
222 
+							}
223 
+							assert.Assert(t, err)
224 
+							assert.Check(t, lines == i, "%d -- %d", lines, i)
225 
+
226 
+							b, err := ioutil.ReadAll(tr)
227 
+							assert.Assert(t, err)
228 
+
229 
+							expectLines := test.data[len(test.data)-i:]
230 
+							assert.Check(t, len(expectLines) == i)
231 
+							expect := strings.Join(expectLines, string(delim)) + string(delim)
232 
+							assert.Check(t, string(b) == expect, "\n%v\n%v", b, []byte(expect))
233 
+						})
234 
+					}
235 
+
236 
+					t.Run("request more lines than available", func(t *testing.T) {
237 
+						t.Parallel()
238 
+
239 
+						r := strings.NewReader(s)
240 
+						tr, lines, err := NewTailReaderWithDelimiter(ctx, r, len(test.data)*2, delim)
241 
+						if len(delim) == 0 {
242 
+							assert.Assert(t, err != nil)
243 
+							assert.Assert(t, lines == 0)
244 
+							return
245 
+						}
246 
+						if len(test.data) == 0 {
247 
+							assert.Assert(t, err == ErrNonPositiveLinesNumber, err)
248 
+							return
249 
+						}
250 
+
251 
+						assert.Assert(t, err)
252 
+						assert.Check(t, lines == len(test.data), "%d -- %d", lines, len(test.data))
253 
+						b, err := ioutil.ReadAll(tr)
254 
+						assert.Assert(t, err)
255 
+						assert.Check(t, bytes.Equal(b, []byte(s)), "\n%v\n%v", b, []byte(s))
256 
+					})
257 
+				})
258 
+			}
259 
+		})
260 
+	}
261 
+	t.Run("truncated last line", func(t *testing.T) {
262 
+		t.Run("more than available", func(t *testing.T) {
263 
+			tail, nLines, err := NewTailReader(ctx, strings.NewReader("a\nb\nextra"), 3)
264 
+			assert.Assert(t, err)
265 
+			assert.Check(t, nLines == 2, nLines)
266 
+
267 
+			rdr := bufio.NewReader(tail)
268 
+			data, _, err := rdr.ReadLine()
269 
+			assert.Assert(t, err)
270 
+			assert.Check(t, string(data) == "a", string(data))
271 
+
272 
+			data, _, err = rdr.ReadLine()
273 
+			assert.Assert(t, err)
274 
+			assert.Check(t, string(data) == "b", string(data))
275 
+
276 
+			_, _, err = rdr.ReadLine()
277 
+			assert.Assert(t, err == io.EOF, err)
278 
+		})
279 
+	})
280 
+	t.Run("truncated last line", func(t *testing.T) {
281 
+		t.Run("exact", func(t *testing.T) {
282 
+			tail, nLines, err := NewTailReader(ctx, strings.NewReader("a\nb\nextra"), 2)
283 
+			assert.Assert(t, err)
284 
+			assert.Check(t, nLines == 2, nLines)
285 
+
286 
+			rdr := bufio.NewReader(tail)
287 
+			data, _, err := rdr.ReadLine()
288 
+			assert.Assert(t, err)
289 
+			assert.Check(t, string(data) == "a", string(data))
290 
+
291 
+			data, _, err = rdr.ReadLine()
292 
+			assert.Assert(t, err)
293 
+			assert.Check(t, string(data) == "b", string(data))
294 
+
295 
+			_, _, err = rdr.ReadLine()
296 
+			assert.Assert(t, err == io.EOF, err)
297 
+		})
298 
+	})
299 
+
300 
+	t.Run("truncated last line", func(t *testing.T) {
301 
+		t.Run("one line", func(t *testing.T) {
302 
+			tail, nLines, err := NewTailReader(ctx, strings.NewReader("a\nb\nextra"), 1)
303 
+			assert.Assert(t, err)
304 
+			assert.Check(t, nLines == 1, nLines)
305 
+
306 
+			rdr := bufio.NewReader(tail)
307 
+			data, _, err := rdr.ReadLine()
308 
+			assert.Assert(t, err)
309 
+			assert.Check(t, string(data) == "b", string(data))
310 
+
311 
+			_, _, err = rdr.ReadLine()
312 
+			assert.Assert(t, err == io.EOF, err)
313 
+		})
314 
+	})
315 
+}