1. docker
  2. Commit 28334c1d829b9ea8ac7e10d45a6db37c3821d5db
Browse code

Merge pull request #28403 from cpuguy83/logging_plugins

Implement plugins for logging drivers

Sebastiaan van Stijn authored on 2017/04/11 04:57:56
Showing 22 changed files
api/types/plugins/logdriver/entry.pb.go
History View file @ 28334c1
1 1 
new file mode 100644
... ... 
@@ -0,0 +1,449 @@
0 
+// Code generated by protoc-gen-gogo.
1 
+// source: entry.proto
2 
+// DO NOT EDIT!
3 
+
4 
+/*
5 
+	Package logdriver is a generated protocol buffer package.
6 
+
7 
+	It is generated from these files:
8 
+		entry.proto
9 
+
10 
+	It has these top-level messages:
11 
+		LogEntry
12 
+*/
13 
+package logdriver
14 
+
15 
+import proto "github.com/gogo/protobuf/proto"
16 
+import fmt "fmt"
17 
+import math "math"
18 
+
19 
+import io "io"
20 
+
21 
+// Reference imports to suppress errors if they are not otherwise used.
22 
+var _ = proto.Marshal
23 
+var _ = fmt.Errorf
24 
+var _ = math.Inf
25 
+
26 
+// This is a compile-time assertion to ensure that this generated file
27 
+// is compatible with the proto package it is being compiled against.
28 
+// A compilation error at this line likely means your copy of the
29 
+// proto package needs to be updated.
30 
+const _ = proto.GoGoProtoPackageIsVersion2 // please upgrade the proto package
31 
+
32 
+type LogEntry struct {
33 
+	Source   string `protobuf:"bytes,1,opt,name=source,proto3" json:"source,omitempty"`
34 
+	TimeNano int64  `protobuf:"varint,2,opt,name=time_nano,json=timeNano,proto3" json:"time_nano,omitempty"`
35 
+	Line     []byte `protobuf:"bytes,3,opt,name=line,proto3" json:"line,omitempty"`
36 
+	Partial  bool   `protobuf:"varint,4,opt,name=partial,proto3" json:"partial,omitempty"`
37 
+}
38 
+
39 
+func (m *LogEntry) Reset()                    { *m = LogEntry{} }
40 
+func (m *LogEntry) String() string            { return proto.CompactTextString(m) }
41 
+func (*LogEntry) ProtoMessage()               {}
42 
+func (*LogEntry) Descriptor() ([]byte, []int) { return fileDescriptorEntry, []int{0} }
43 
+
44 
+func (m *LogEntry) GetSource() string {
45 
+	if m != nil {
46 
+		return m.Source
47 
+	}
48 
+	return ""
49 
+}
50 
+
51 
+func (m *LogEntry) GetTimeNano() int64 {
52 
+	if m != nil {
53 
+		return m.TimeNano
54 
+	}
55 
+	return 0
56 
+}
57 
+
58 
+func (m *LogEntry) GetLine() []byte {
59 
+	if m != nil {
60 
+		return m.Line
61 
+	}
62 
+	return nil
63 
+}
64 
+
65 
+func (m *LogEntry) GetPartial() bool {
66 
+	if m != nil {
67 
+		return m.Partial
68 
+	}
69 
+	return false
70 
+}
71 
+
72 
+func init() {
73 
+	proto.RegisterType((*LogEntry)(nil), "LogEntry")
74 
+}
75 
+func (m *LogEntry) Marshal() (dAtA []byte, err error) {
76 
+	size := m.Size()
77 
+	dAtA = make([]byte, size)
78 
+	n, err := m.MarshalTo(dAtA)
79 
+	if err != nil {
80 
+		return nil, err
81 
+	}
82 
+	return dAtA[:n], nil
83 
+}
84 
+
85 
+func (m *LogEntry) MarshalTo(dAtA []byte) (int, error) {
86 
+	var i int
87 
+	_ = i
88 
+	var l int
89 
+	_ = l
90 
+	if len(m.Source) > 0 {
91 
+		dAtA[i] = 0xa
92 
+		i++
93 
+		i = encodeVarintEntry(dAtA, i, uint64(len(m.Source)))
94 
+		i += copy(dAtA[i:], m.Source)
95 
+	}
96 
+	if m.TimeNano != 0 {
97 
+		dAtA[i] = 0x10
98 
+		i++
99 
+		i = encodeVarintEntry(dAtA, i, uint64(m.TimeNano))
100 
+	}
101 
+	if len(m.Line) > 0 {
102 
+		dAtA[i] = 0x1a
103 
+		i++
104 
+		i = encodeVarintEntry(dAtA, i, uint64(len(m.Line)))
105 
+		i += copy(dAtA[i:], m.Line)
106 
+	}
107 
+	if m.Partial {
108 
+		dAtA[i] = 0x20
109 
+		i++
110 
+		if m.Partial {
111 
+			dAtA[i] = 1
112 
+		} else {
113 
+			dAtA[i] = 0
114 
+		}
115 
+		i++
116 
+	}
117 
+	return i, nil
118 
+}
119 
+
120 
+func encodeFixed64Entry(dAtA []byte, offset int, v uint64) int {
121 
+	dAtA[offset] = uint8(v)
122 
+	dAtA[offset+1] = uint8(v >> 8)
123 
+	dAtA[offset+2] = uint8(v >> 16)
124 
+	dAtA[offset+3] = uint8(v >> 24)
125 
+	dAtA[offset+4] = uint8(v >> 32)
126 
+	dAtA[offset+5] = uint8(v >> 40)
127 
+	dAtA[offset+6] = uint8(v >> 48)
128 
+	dAtA[offset+7] = uint8(v >> 56)
129 
+	return offset + 8
130 
+}
131 
+func encodeFixed32Entry(dAtA []byte, offset int, v uint32) int {
132 
+	dAtA[offset] = uint8(v)
133 
+	dAtA[offset+1] = uint8(v >> 8)
134 
+	dAtA[offset+2] = uint8(v >> 16)
135 
+	dAtA[offset+3] = uint8(v >> 24)
136 
+	return offset + 4
137 
+}
138 
+func encodeVarintEntry(dAtA []byte, offset int, v uint64) int {
139 
+	for v >= 1<<7 {
140 
+		dAtA[offset] = uint8(v&0x7f | 0x80)
141 
+		v >>= 7
142 
+		offset++
143 
+	}
144 
+	dAtA[offset] = uint8(v)
145 
+	return offset + 1
146 
+}
147 
+func (m *LogEntry) Size() (n int) {
148 
+	var l int
149 
+	_ = l
150 
+	l = len(m.Source)
151 
+	if l > 0 {
152 
+		n += 1 + l + sovEntry(uint64(l))
153 
+	}
154 
+	if m.TimeNano != 0 {
155 
+		n += 1 + sovEntry(uint64(m.TimeNano))
156 
+	}
157 
+	l = len(m.Line)
158 
+	if l > 0 {
159 
+		n += 1 + l + sovEntry(uint64(l))
160 
+	}
161 
+	if m.Partial {
162 
+		n += 2
163 
+	}
164 
+	return n
165 
+}
166 
+
167 
+func sovEntry(x uint64) (n int) {
168 
+	for {
169 
+		n++
170 
+		x >>= 7
171 
+		if x == 0 {
172 
+			break
173 
+		}
174 
+	}
175 
+	return n
176 
+}
177 
+func sozEntry(x uint64) (n int) {
178 
+	return sovEntry(uint64((x << 1) ^ uint64((int64(x) >> 63))))
179 
+}
180 
+func (m *LogEntry) Unmarshal(dAtA []byte) error {
181 
+	l := len(dAtA)
182 
+	iNdEx := 0
183 
+	for iNdEx < l {
184 
+		preIndex := iNdEx
185 
+		var wire uint64
186 
+		for shift := uint(0); ; shift += 7 {
187 
+			if shift >= 64 {
188 
+				return ErrIntOverflowEntry
189 
+			}
190 
+			if iNdEx >= l {
191 
+				return io.ErrUnexpectedEOF
192 
+			}
193 
+			b := dAtA[iNdEx]
194 
+			iNdEx++
195 
+			wire |= (uint64(b) & 0x7F) << shift
196 
+			if b < 0x80 {
197 
+				break
198 
+			}
199 
+		}
200 
+		fieldNum := int32(wire >> 3)
201 
+		wireType := int(wire & 0x7)
202 
+		if wireType == 4 {
203 
+			return fmt.Errorf("proto: LogEntry: wiretype end group for non-group")
204 
+		}
205 
+		if fieldNum <= 0 {
206 
+			return fmt.Errorf("proto: LogEntry: illegal tag %d (wire type %d)", fieldNum, wire)
207 
+		}
208 
+		switch fieldNum {
209 
+		case 1:
210 
+			if wireType != 2 {
211 
+				return fmt.Errorf("proto: wrong wireType = %d for field Source", wireType)
212 
+			}
213 
+			var stringLen uint64
214 
+			for shift := uint(0); ; shift += 7 {
215 
+				if shift >= 64 {
216 
+					return ErrIntOverflowEntry
217 
+				}
218 
+				if iNdEx >= l {
219 
+					return io.ErrUnexpectedEOF
220 
+				}
221 
+				b := dAtA[iNdEx]
222 
+				iNdEx++
223 
+				stringLen |= (uint64(b) & 0x7F) << shift
224 
+				if b < 0x80 {
225 
+					break
226 
+				}
227 
+			}
228 
+			intStringLen := int(stringLen)
229 
+			if intStringLen < 0 {
230 
+				return ErrInvalidLengthEntry
231 
+			}
232 
+			postIndex := iNdEx + intStringLen
233 
+			if postIndex > l {
234 
+				return io.ErrUnexpectedEOF
235 
+			}
236 
+			m.Source = string(dAtA[iNdEx:postIndex])
237 
+			iNdEx = postIndex
238 
+		case 2:
239 
+			if wireType != 0 {
240 
+				return fmt.Errorf("proto: wrong wireType = %d for field TimeNano", wireType)
241 
+			}
242 
+			m.TimeNano = 0
243 
+			for shift := uint(0); ; shift += 7 {
244 
+				if shift >= 64 {
245 
+					return ErrIntOverflowEntry
246 
+				}
247 
+				if iNdEx >= l {
248 
+					return io.ErrUnexpectedEOF
249 
+				}
250 
+				b := dAtA[iNdEx]
251 
+				iNdEx++
252 
+				m.TimeNano |= (int64(b) & 0x7F) << shift
253 
+				if b < 0x80 {
254 
+					break
255 
+				}
256 
+			}
257 
+		case 3:
258 
+			if wireType != 2 {
259 
+				return fmt.Errorf("proto: wrong wireType = %d for field Line", wireType)
260 
+			}
261 
+			var byteLen int
262 
+			for shift := uint(0); ; shift += 7 {
263 
+				if shift >= 64 {
264 
+					return ErrIntOverflowEntry
265 
+				}
266 
+				if iNdEx >= l {
267 
+					return io.ErrUnexpectedEOF
268 
+				}
269 
+				b := dAtA[iNdEx]
270 
+				iNdEx++
271 
+				byteLen |= (int(b) & 0x7F) << shift
272 
+				if b < 0x80 {
273 
+					break
274 
+				}
275 
+			}
276 
+			if byteLen < 0 {
277 
+				return ErrInvalidLengthEntry
278 
+			}
279 
+			postIndex := iNdEx + byteLen
280 
+			if postIndex > l {
281 
+				return io.ErrUnexpectedEOF
282 
+			}
283 
+			m.Line = append(m.Line[:0], dAtA[iNdEx:postIndex]...)
284 
+			if m.Line == nil {
285 
+				m.Line = []byte{}
286 
+			}
287 
+			iNdEx = postIndex
288 
+		case 4:
289 
+			if wireType != 0 {
290 
+				return fmt.Errorf("proto: wrong wireType = %d for field Partial", wireType)
291 
+			}
292 
+			var v int
293 
+			for shift := uint(0); ; shift += 7 {
294 
+				if shift >= 64 {
295 
+					return ErrIntOverflowEntry
296 
+				}
297 
+				if iNdEx >= l {
298 
+					return io.ErrUnexpectedEOF
299 
+				}
300 
+				b := dAtA[iNdEx]
301 
+				iNdEx++
302 
+				v |= (int(b) & 0x7F) << shift
303 
+				if b < 0x80 {
304 
+					break
305 
+				}
306 
+			}
307 
+			m.Partial = bool(v != 0)
308 
+		default:
309 
+			iNdEx = preIndex
310 
+			skippy, err := skipEntry(dAtA[iNdEx:])
311 
+			if err != nil {
312 
+				return err
313 
+			}
314 
+			if skippy < 0 {
315 
+				return ErrInvalidLengthEntry
316 
+			}
317 
+			if (iNdEx + skippy) > l {
318 
+				return io.ErrUnexpectedEOF
319 
+			}
320 
+			iNdEx += skippy
321 
+		}
322 
+	}
323 
+
324 
+	if iNdEx > l {
325 
+		return io.ErrUnexpectedEOF
326 
+	}
327 
+	return nil
328 
+}
329 
+func skipEntry(dAtA []byte) (n int, err error) {
330 
+	l := len(dAtA)
331 
+	iNdEx := 0
332 
+	for iNdEx < l {
333 
+		var wire uint64
334 
+		for shift := uint(0); ; shift += 7 {
335 
+			if shift >= 64 {
336 
+				return 0, ErrIntOverflowEntry
337 
+			}
338 
+			if iNdEx >= l {
339 
+				return 0, io.ErrUnexpectedEOF
340 
+			}
341 
+			b := dAtA[iNdEx]
342 
+			iNdEx++
343 
+			wire |= (uint64(b) & 0x7F) << shift
344 
+			if b < 0x80 {
345 
+				break
346 
+			}
347 
+		}
348 
+		wireType := int(wire & 0x7)
349 
+		switch wireType {
350 
+		case 0:
351 
+			for shift := uint(0); ; shift += 7 {
352 
+				if shift >= 64 {
353 
+					return 0, ErrIntOverflowEntry
354 
+				}
355 
+				if iNdEx >= l {
356 
+					return 0, io.ErrUnexpectedEOF
357 
+				}
358 
+				iNdEx++
359 
+				if dAtA[iNdEx-1] < 0x80 {
360 
+					break
361 
+				}
362 
+			}
363 
+			return iNdEx, nil
364 
+		case 1:
365 
+			iNdEx += 8
366 
+			return iNdEx, nil
367 
+		case 2:
368 
+			var length int
369 
+			for shift := uint(0); ; shift += 7 {
370 
+				if shift >= 64 {
371 
+					return 0, ErrIntOverflowEntry
372 
+				}
373 
+				if iNdEx >= l {
374 
+					return 0, io.ErrUnexpectedEOF
375 
+				}
376 
+				b := dAtA[iNdEx]
377 
+				iNdEx++
378 
+				length |= (int(b) & 0x7F) << shift
379 
+				if b < 0x80 {
380 
+					break
381 
+				}
382 
+			}
383 
+			iNdEx += length
384 
+			if length < 0 {
385 
+				return 0, ErrInvalidLengthEntry
386 
+			}
387 
+			return iNdEx, nil
388 
+		case 3:
389 
+			for {
390 
+				var innerWire uint64
391 
+				var start int = iNdEx
392 
+				for shift := uint(0); ; shift += 7 {
393 
+					if shift >= 64 {
394 
+						return 0, ErrIntOverflowEntry
395 
+					}
396 
+					if iNdEx >= l {
397 
+						return 0, io.ErrUnexpectedEOF
398 
+					}
399 
+					b := dAtA[iNdEx]
400 
+					iNdEx++
401 
+					innerWire |= (uint64(b) & 0x7F) << shift
402 
+					if b < 0x80 {
403 
+						break
404 
+					}
405 
+				}
406 
+				innerWireType := int(innerWire & 0x7)
407 
+				if innerWireType == 4 {
408 
+					break
409 
+				}
410 
+				next, err := skipEntry(dAtA[start:])
411 
+				if err != nil {
412 
+					return 0, err
413 
+				}
414 
+				iNdEx = start + next
415 
+			}
416 
+			return iNdEx, nil
417 
+		case 4:
418 
+			return iNdEx, nil
419 
+		case 5:
420 
+			iNdEx += 4
421 
+			return iNdEx, nil
422 
+		default:
423 
+			return 0, fmt.Errorf("proto: illegal wireType %d", wireType)
424 
+		}
425 
+	}
426 
+	panic("unreachable")
427 
+}
428 
+
429 
+var (
430 
+	ErrInvalidLengthEntry = fmt.Errorf("proto: negative length found during unmarshaling")
431 
+	ErrIntOverflowEntry   = fmt.Errorf("proto: integer overflow")
432 
+)
433 
+
434 
+func init() { proto.RegisterFile("entry.proto", fileDescriptorEntry) }
435 
+
436 
+var fileDescriptorEntry = []byte{
437 
+	// 149 bytes of a gzipped FileDescriptorProto
438 
+	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x09, 0x6e, 0x88, 0x02, 0xff, 0xe2, 0xe2, 0x4e, 0xcd, 0x2b, 0x29,
439 
+	0xaa, 0xd4, 0x2b, 0x28, 0xca, 0x2f, 0xc9, 0x57, 0xca, 0xe5, 0xe2, 0xf0, 0xc9, 0x4f, 0x77, 0x05,
440 
+	0x89, 0x08, 0x89, 0x71, 0xb1, 0x15, 0xe7, 0x97, 0x16, 0x25, 0xa7, 0x4a, 0x30, 0x2a, 0x30, 0x6a,
441 
+	0x70, 0x06, 0x41, 0x79, 0x42, 0xd2, 0x5c, 0x9c, 0x25, 0x99, 0xb9, 0xa9, 0xf1, 0x79, 0x89, 0x79,
442 
+	0xf9, 0x12, 0x4c, 0x0a, 0x8c, 0x1a, 0xcc, 0x41, 0x1c, 0x20, 0x01, 0xbf, 0xc4, 0xbc, 0x7c, 0x21,
443 
+	0x21, 0x2e, 0x96, 0x9c, 0xcc, 0xbc, 0x54, 0x09, 0x66, 0x05, 0x46, 0x0d, 0x9e, 0x20, 0x30, 0x5b,
444 
+	0x48, 0x82, 0x8b, 0xbd, 0x20, 0xb1, 0xa8, 0x24, 0x33, 0x31, 0x47, 0x82, 0x45, 0x81, 0x51, 0x83,
445 
+	0x23, 0x08, 0xc6, 0x75, 0xe2, 0x39, 0xf1, 0x48, 0x8e, 0xf1, 0xc2, 0x23, 0x39, 0xc6, 0x07, 0x8f,
446 
+	0xe4, 0x18, 0x93, 0xd8, 0xc0, 0x6e, 0x30, 0x06, 0x04, 0x00, 0x00, 0xff, 0xff, 0x2d, 0x24, 0x5a,
447 
+	0xd4, 0x92, 0x00, 0x00, 0x00,
448 
+}
api/types/plugins/logdriver/entry.proto
History View file @ 28334c1
0 449 
new file mode 100644
... ... 
@@ -0,0 +1,8 @@
0 
+syntax = "proto3";
1 
+
2 
+message LogEntry {
3 
+	string source = 1;
4 
+	int64 time_nano = 2;
5 
+	bytes line = 3;
6 
+	bool partial = 4;
7 
+}
api/types/plugins/logdriver/gen.go
History View file @ 28334c1
0 8 
new file mode 100644
... ... 
@@ -0,0 +1,3 @@
0 
+//go:generate protoc --gogofast_out=import_path=github.com/docker/docker/api/types/plugins/logdriver:. entry.proto
1 
+
2 
+package logdriver
api/types/plugins/logdriver/io.go
History View file @ 28334c1
0 3 
new file mode 100644
... ... 
@@ -0,0 +1,87 @@
0 
+package logdriver
1 
+
2 
+import (
3 
+	"encoding/binary"
4 
+	"io"
5 
+)
6 
+
7 
+const binaryEncodeLen = 4
8 
+
9 
+// LogEntryEncoder encodes a LogEntry to a protobuf stream
10 
+// The stream should look like:
11 
+//
12 
+// [uint32 binary encoded message size][protobuf message]
13 
+//
14 
+// To decode an entry, read the first 4 bytes to get the size of the entry,
15 
+// then read `size` bytes from the stream.
16 
+type LogEntryEncoder interface {
17 
+	Encode(*LogEntry) error
18 
+}
19 
+
20 
+// NewLogEntryEncoder creates a protobuf stream encoder for log entries.
21 
+// This is used to write out  log entries to a stream.
22 
+func NewLogEntryEncoder(w io.Writer) LogEntryEncoder {
23 
+	return &logEntryEncoder{
24 
+		w:   w,
25 
+		buf: make([]byte, 1024),
26 
+	}
27 
+}
28 
+
29 
+type logEntryEncoder struct {
30 
+	buf []byte
31 
+	w   io.Writer
32 
+}
33 
+
34 
+func (e *logEntryEncoder) Encode(l *LogEntry) error {
35 
+	n := l.Size()
36 
+
37 
+	total := n + binaryEncodeLen
38 
+	if total > len(e.buf) {
39 
+		e.buf = make([]byte, total)
40 
+	}
41 
+	binary.BigEndian.PutUint32(e.buf, uint32(n))
42 
+
43 
+	if _, err := l.MarshalTo(e.buf[binaryEncodeLen:]); err != nil {
44 
+		return err
45 
+	}
46 
+	_, err := e.w.Write(e.buf[:total])
47 
+	return err
48 
+}
49 
+
50 
+// LogEntryDecoder decodes log entries from a stream
51 
+// It is expected that the wire format is as defined by LogEntryEncoder.
52 
+type LogEntryDecoder interface {
53 
+	Decode(*LogEntry) error
54 
+}
55 
+
56 
+// NewLogEntryDecoder creates a new stream decoder for log entries
57 
+func NewLogEntryDecoder(r io.Reader) LogEntryDecoder {
58 
+	return &logEntryDecoder{
59 
+		lenBuf: make([]byte, binaryEncodeLen),
60 
+		buf:    make([]byte, 1024),
61 
+		r:      r,
62 
+	}
63 
+}
64 
+
65 
+type logEntryDecoder struct {
66 
+	r      io.Reader
67 
+	lenBuf []byte
68 
+	buf    []byte
69 
+}
70 
+
71 
+func (d *logEntryDecoder) Decode(l *LogEntry) error {
72 
+	_, err := io.ReadFull(d.r, d.lenBuf)
73 
+	if err != nil {
74 
+		return err
75 
+	}
76 
+
77 
+	size := int(binary.BigEndian.Uint32(d.lenBuf))
78 
+	if len(d.buf) < size {
79 
+		d.buf = make([]byte, size)
80 
+	}
81 
+
82 
+	if _, err := io.ReadFull(d.r, d.buf[:size]); err != nil {
83 
+		return err
84 
+	}
85 
+	return l.Unmarshal(d.buf[:size])
86 
+}
daemon/daemon.go
History View file @ 28334c1
... ... 
@@ -27,6 +27,7 @@ import (
27 27 
 	"github.com/docker/docker/daemon/discovery"
28 28 
 	"github.com/docker/docker/daemon/events"
29 29 
 	"github.com/docker/docker/daemon/exec"
30 
+	"github.com/docker/docker/daemon/logger"
30 31 
 	// register graph drivers
31 32 
 	_ "github.com/docker/docker/daemon/graphdriver/register"
32 33 
 	"github.com/docker/docker/daemon/initlayer"
... ... 
@@ -586,6 +587,7 @@ func NewDaemon(config *config.Config, registryService registry.Service, containe
586 586
587 587 
 	d.RegistryService = registryService
588 588 
 	d.PluginStore = pluginStore
589 
+	logger.RegisterPluginGetter(d.PluginStore)
589 590
590 591 
 	// Plugin system initialization should happen before restore. Do not change order.
591 592 
 	d.pluginManager, err = plugin.NewManager(plugin.ManagerConfig{
daemon/logger/adapter.go
History View file @ 28334c1
592 593 
new file mode 100644
... ... 
@@ -0,0 +1,135 @@
0 
+package logger
1 
+
2 
+import (
3 
+	"io"
4 
+	"os"
5 
+	"sync"
6 
+	"time"
7 
+
8 
+	"github.com/Sirupsen/logrus"
9 
+	"github.com/docker/docker/api/types/plugins/logdriver"
10 
+	"github.com/docker/docker/pkg/plugingetter"
11 
+	"github.com/pkg/errors"
12 
+)
13 
+
14 
+// pluginAdapter takes a plugin and implements the Logger interface for logger
15 
+// instances
16 
+type pluginAdapter struct {
17 
+	driverName   string
18 
+	id           string
19 
+	plugin       logPlugin
20 
+	fifoPath     string
21 
+	capabilities Capability
22 
+	logInfo      Info
23 
+
24 
+	// synchronize access to the log stream and shared buffer
25 
+	mu     sync.Mutex
26 
+	enc    logdriver.LogEntryEncoder
27 
+	stream io.WriteCloser
28 
+	// buf is shared for each `Log()` call to reduce allocations.
29 
+	// buf must be protected by mutex
30 
+	buf logdriver.LogEntry
31 
+}
32 
+
33 
+func (a *pluginAdapter) Log(msg *Message) error {
34 
+	a.mu.Lock()
35 
+
36 
+	a.buf.Line = msg.Line
37 
+	a.buf.TimeNano = msg.Timestamp.UnixNano()
38 
+	a.buf.Partial = msg.Partial
39 
+	a.buf.Source = msg.Source
40 
+
41 
+	err := a.enc.Encode(&a.buf)
42 
+	a.buf.Reset()
43 
+
44 
+	a.mu.Unlock()
45 
+
46 
+	PutMessage(msg)
47 
+	return err
48 
+}
49 
+
50 
+func (a *pluginAdapter) Name() string {
51 
+	return a.driverName
52 
+}
53 
+
54 
+func (a *pluginAdapter) Close() error {
55 
+	a.mu.Lock()
56 
+	defer a.mu.Unlock()
57 
+
58 
+	if err := a.plugin.StopLogging(a.fifoPath); err != nil {
59 
+		return err
60 
+	}
61 
+
62 
+	if err := a.stream.Close(); err != nil {
63 
+		logrus.WithError(err).Error("error closing plugin fifo")
64 
+	}
65 
+	if err := os.Remove(a.fifoPath); err != nil && !os.IsNotExist(err) {
66 
+		logrus.WithError(err).Error("error cleaning up plugin fifo")
67 
+	}
68 
+
69 
+	// may be nil, especially for unit tests
70 
+	if pluginGetter != nil {
71 
+		pluginGetter.Get(a.Name(), extName, plugingetter.Release)
72 
+	}
73 
+	return nil
74 
+}
75 
+
76 
+type pluginAdapterWithRead struct {
77 
+	*pluginAdapter
78 
+}
79 
+
80 
+func (a *pluginAdapterWithRead) ReadLogs(config ReadConfig) *LogWatcher {
81 
+	watcher := NewLogWatcher()
82 
+
83 
+	go func() {
84 
+		defer close(watcher.Msg)
85 
+		stream, err := a.plugin.ReadLogs(a.logInfo, config)
86 
+		if err != nil {
87 
+			watcher.Err <- errors.Wrap(err, "error getting log reader")
88 
+			return
89 
+		}
90 
+		defer stream.Close()
91 
+
92 
+		dec := logdriver.NewLogEntryDecoder(stream)
93 
+		for {
94 
+			select {
95 
+			case <-watcher.WatchClose():
96 
+				return
97 
+			default:
98 
+			}
99 
+
100 
+			var buf logdriver.LogEntry
101 
+			if err := dec.Decode(&buf); err != nil {
102 
+				if err == io.EOF {
103 
+					return
104 
+				}
105 
+				select {
106 
+				case watcher.Err <- errors.Wrap(err, "error decoding log message"):
107 
+				case <-watcher.WatchClose():
108 
+				}
109 
+				return
110 
+			}
111 
+
112 
+			msg := &Message{
113 
+				Timestamp: time.Unix(0, buf.TimeNano),
114 
+				Line:      buf.Line,
115 
+				Source:    buf.Source,
116 
+			}
117 
+
118 
+			// plugin should handle this, but check just in case
119 
+			if !config.Since.IsZero() && msg.Timestamp.Before(config.Since) {
120 
+				continue
121 
+			}
122 
+
123 
+			select {
124 
+			case watcher.Msg <- msg:
125 
+			case <-watcher.WatchClose():
126 
+				// make sure the message we consumed is sent
127 
+				watcher.Msg <- msg
128 
+				return
129 
+			}
130 
+		}
131 
+	}()
132 
+
133 
+	return watcher
134 
+}
daemon/logger/adapter_test.go
History View file @ 28334c1
0 135 
new file mode 100644
... ... 
@@ -0,0 +1,208 @@
0 
+package logger
1 
+
2 
+import (
3 
+	"bytes"
4 
+	"encoding/binary"
5 
+	"io"
6 
+	"io/ioutil"
7 
+	"os"
8 
+	"runtime"
9 
+	"testing"
10 
+	"time"
11 
+
12 
+	"github.com/docker/docker/api/types/plugins/logdriver"
13 
+	protoio "github.com/gogo/protobuf/io"
14 
+)
15 
+
16 
+// mockLoggingPlugin implements the loggingPlugin interface for testing purposes
17 
+// it only supports a single log stream
18 
+type mockLoggingPlugin struct {
19 
+	inStream io.ReadCloser
20 
+	f        *os.File
21 
+	closed   chan struct{}
22 
+	t        *testing.T
23 
+}
24 
+
25 
+func (l *mockLoggingPlugin) StartLogging(file string, info Info) error {
26 
+	go func() {
27 
+		io.Copy(l.f, l.inStream)
28 
+		close(l.closed)
29 
+	}()
30 
+	return nil
31 
+}
32 
+
33 
+func (l *mockLoggingPlugin) StopLogging(file string) error {
34 
+	l.inStream.Close()
35 
+	l.f.Close()
36 
+	os.Remove(l.f.Name())
37 
+	return nil
38 
+}
39 
+
40 
+func (l *mockLoggingPlugin) Capabilities() (cap Capability, err error) {
41 
+	return Capability{ReadLogs: true}, nil
42 
+}
43 
+
44 
+func (l *mockLoggingPlugin) ReadLogs(info Info, config ReadConfig) (io.ReadCloser, error) {
45 
+	r, w := io.Pipe()
46 
+	f, err := os.Open(l.f.Name())
47 
+	if err != nil {
48 
+		return nil, err
49 
+	}
50 
+	go func() {
51 
+		defer f.Close()
52 
+		dec := protoio.NewUint32DelimitedReader(f, binary.BigEndian, 1e6)
53 
+		enc := logdriver.NewLogEntryEncoder(w)
54 
+
55 
+		for {
56 
+			select {
57 
+			case <-l.closed:
58 
+				w.Close()
59 
+				return
60 
+			default:
61 
+			}
62 
+
63 
+			var msg logdriver.LogEntry
64 
+			if err := dec.ReadMsg(&msg); err != nil {
65 
+				if err == io.EOF {
66 
+					if !config.Follow {
67 
+						w.Close()
68 
+						return
69 
+					}
70 
+					dec = protoio.NewUint32DelimitedReader(f, binary.BigEndian, 1e6)
71 
+					continue
72 
+				}
73 
+
74 
+				l.t.Fatal(err)
75 
+				continue
76 
+			}
77 
+
78 
+			if err := enc.Encode(&msg); err != nil {
79 
+				w.CloseWithError(err)
80 
+				return
81 
+			}
82 
+		}
83 
+	}()
84 
+
85 
+	return r, nil
86 
+}
87 
+
88 
+func newMockPluginAdapter(t *testing.T) Logger {
89 
+	r, w := io.Pipe()
90 
+	f, err := ioutil.TempFile("", "mock-plugin-adapter")
91 
+	if err != nil {
92 
+		t.Fatal(err)
93 
+	}
94 
+	enc := logdriver.NewLogEntryEncoder(w)
95 
+	a := &pluginAdapterWithRead{
96 
+		&pluginAdapter{
97 
+			plugin: &mockLoggingPlugin{
98 
+				inStream: r,
99 
+				f:        f,
100 
+				closed:   make(chan struct{}),
101 
+				t:        t,
102 
+			},
103 
+			stream: w,
104 
+			enc:    enc,
105 
+		},
106 
+	}
107 
+	a.plugin.StartLogging("", Info{})
108 
+	return a
109 
+}
110 
+
111 
+func TestAdapterReadLogs(t *testing.T) {
112 
+	l := newMockPluginAdapter(t)
113 
+
114 
+	testMsg := []Message{
115 
+		{Line: []byte("Are you the keymaker?"), Timestamp: time.Now()},
116 
+		{Line: []byte("Follow the white rabbit"), Timestamp: time.Now()},
117 
+	}
118 
+	for _, msg := range testMsg {
119 
+		m := msg.copy()
120 
+		if err := l.Log(m); err != nil {
121 
+			t.Fatal(err)
122 
+		}
123 
+	}
124 
+
125 
+	lr, ok := l.(LogReader)
126 
+	if !ok {
127 
+		t.Fatal("expected log reader")
128 
+	}
129 
+
130 
+	lw := lr.ReadLogs(ReadConfig{})
131 
+
132 
+	for _, x := range testMsg {
133 
+		select {
134 
+		case msg := <-lw.Msg:
135 
+			testMessageEqual(t, &x, msg)
136 
+		case <-time.After(10 * time.Millisecond):
137 
+			t.Fatal("timeout reading logs")
138 
+		}
139 
+	}
140 
+
141 
+	select {
142 
+	case _, ok := <-lw.Msg:
143 
+		if ok {
144 
+			t.Fatal("expected message channel to be closed")
145 
+		}
146 
+	case <-time.After(10 * time.Second):
147 
+		t.Fatal("timeout waiting for message channel to close")
148 
+
149 
+	}
150 
+	lw.Close()
151 
+
152 
+	lw = lr.ReadLogs(ReadConfig{Follow: true})
153 
+	for _, x := range testMsg {
154 
+		select {
155 
+		case msg := <-lw.Msg:
156 
+			testMessageEqual(t, &x, msg)
157 
+		case <-time.After(10 * time.Second):
158 
+			t.Fatal("timeout reading logs")
159 
+		}
160 
+	}
161 
+
162 
+	x := Message{Line: []byte("Too infinity and beyond!"), Timestamp: time.Now()}
163 
+
164 
+	if err := l.Log(x.copy()); err != nil {
165 
+		t.Fatal(err)
166 
+	}
167 
+
168 
+	select {
169 
+	case msg, ok := <-lw.Msg:
170 
+		if !ok {
171 
+			t.Fatal("message channel unexpectedly closed")
172 
+		}
173 
+		testMessageEqual(t, &x, msg)
174 
+	case <-time.After(10 * time.Second):
175 
+		t.Fatal("timeout reading logs")
176 
+	}
177 
+
178 
+	l.Close()
179 
+	select {
180 
+	case msg, ok := <-lw.Msg:
181 
+		if ok {
182 
+			t.Fatal("expected message channel to be closed")
183 
+		}
184 
+		if msg != nil {
185 
+			t.Fatal("expected nil message")
186 
+		}
187 
+	case <-time.After(10 * time.Second):
188 
+		t.Fatal("timeout waiting for logger to close")
189 
+	}
190 
+}
191 
+
192 
+func testMessageEqual(t *testing.T, a, b *Message) {
193 
+	_, _, n, _ := runtime.Caller(1)
194 
+	errFmt := "line %d: expected same messages:\nwant: %+v\nhave: %+v"
195 
+
196 
+	if !bytes.Equal(a.Line, b.Line) {
197 
+		t.Fatalf(errFmt, n, *a, *b)
198 
+	}
199 
+
200 
+	if a.Timestamp.UnixNano() != b.Timestamp.UnixNano() {
201 
+		t.Fatalf(errFmt, n, *a, *b)
202 
+	}
203 
+
204 
+	if a.Source != b.Source {
205 
+		t.Fatalf(errFmt, n, *a, *b)
206 
+	}
207 
+}
daemon/logger/factory.go
History View file @ 28334c1
... ... 
@@ -5,6 +5,7 @@ import (
5 5 
 	"sync"
6 6
7 7 
 	containertypes "github.com/docker/docker/api/types/container"
8 
+	"github.com/docker/docker/pkg/plugingetter"
8 9 
 	units "github.com/docker/go-units"
9 10 
 	"github.com/pkg/errors"
10 11 
 )
... ... 
@@ -37,6 +38,13 @@ func (lf *logdriverFactory) driverRegistered(name string) bool {
37 37 
 	lf.m.Lock()
38 38 
 	_, ok := lf.registry[name]
39 39 
 	lf.m.Unlock()
40 
+	if !ok {
41 
+		if pluginGetter != nil { // this can be nil when the init functions are running
42 
+			if l, _ := getPlugin(name, plugingetter.Lookup); l != nil {
43 
+				return true
44 
+			}
45 
+		}
46 
+	}
40 47 
 	return ok
41 48 
 }
42 49
... ... 
@@ -56,10 +64,12 @@ func (lf *logdriverFactory) get(name string) (Creator, error) {
56 56 
 	defer lf.m.Unlock()
57 57
58 58 
 	c, ok := lf.registry[name]
59 
-	if !ok {
60 
-		return c, fmt.Errorf("logger: no log driver named '%s' is registered", name)
59 
+	if ok {
60 
+		return c, nil
61 61 
 	}
62 
-	return c, nil
62 
+
63 
+	c, err := getPlugin(name, plugingetter.Acquire)
64 
+	return c, errors.Wrapf(err, "logger: no log driver named '%s' is registered", name)
63 65 
 }
64 66
65 67 
 func (lf *logdriverFactory) getLogOptValidator(name string) LogOptValidator {
daemon/logger/logger.go
History View file @ 28334c1
... ... 
@@ -126,3 +126,11 @@ func (w *LogWatcher) Close() {
126 126 
 func (w *LogWatcher) WatchClose() <-chan struct{} {
127 127 
 	return w.closeNotifier
128 128 
 }
129 
+
130 
+// Capability defines the list of capabilties that a driver can implement
131 
+// These capabilities are not required to be a logging driver, however do
132 
+// determine how a logging driver can be used
133 
+type Capability struct {
134 
+	// Determines if a log driver can read back logs
135 
+	ReadLogs bool
136 
+}
daemon/logger/logger_test.go
History View file @ 28334c1
129 137 
new file mode 100644
... ... 
@@ -0,0 +1,19 @@
0 
+package logger
1 
+
2 
+func (m *Message) copy() *Message {
3 
+	msg := &Message{
4 
+		Source:    m.Source,
5 
+		Partial:   m.Partial,
6 
+		Timestamp: m.Timestamp,
7 
+	}
8 
+
9 
+	if m.Attrs != nil {
10 
+		msg.Attrs = make(map[string]string, len(m.Attrs))
11 
+		for k, v := range m.Attrs {
12 
+			msg.Attrs[k] = v
13 
+		}
14 
+	}
15 
+
16 
+	msg.Line = append(make([]byte, 0, len(m.Line)), m.Line...)
17 
+	return msg
18 
+}
daemon/logger/plugin.go
History View file @ 28334c1
0 19 
new file mode 100644
... ... 
@@ -0,0 +1,89 @@
0 
+package logger
1 
+
2 
+import (
3 
+	"fmt"
4 
+	"io"
5 
+	"os"
6 
+	"path/filepath"
7 
+	"strings"
8 
+
9 
+	"github.com/docker/docker/api/types/plugins/logdriver"
10 
+	getter "github.com/docker/docker/pkg/plugingetter"
11 
+	"github.com/docker/docker/pkg/stringid"
12 
+	"github.com/pkg/errors"
13 
+)
14 
+
15 
+var pluginGetter getter.PluginGetter
16 
+
17 
+const extName = "LogDriver"
18 
+
19 
+// logPlugin defines the available functions that logging plugins must implement.
20 
+type logPlugin interface {
21 
+	StartLogging(streamPath string, info Info) (err error)
22 
+	StopLogging(streamPath string) (err error)
23 
+	Capabilities() (cap Capability, err error)
24 
+	ReadLogs(info Info, config ReadConfig) (stream io.ReadCloser, err error)
25 
+}
26 
+
27 
+// RegisterPluginGetter sets the plugingetter
28 
+func RegisterPluginGetter(plugingetter getter.PluginGetter) {
29 
+	pluginGetter = plugingetter
30 
+}
31 
+
32 
+// GetDriver returns a logging driver by its name.
33 
+// If the driver is empty, it looks for the local driver.
34 
+func getPlugin(name string, mode int) (Creator, error) {
35 
+	p, err := pluginGetter.Get(name, extName, mode)
36 
+	if err != nil {
37 
+		return nil, fmt.Errorf("error looking up logging plugin %s: %v", name, err)
38 
+	}
39 
+
40 
+	d := &logPluginProxy{p.Client()}
41 
+	return makePluginCreator(name, d, p.BasePath()), nil
42 
+}
43 
+
44 
+func makePluginCreator(name string, l *logPluginProxy, basePath string) Creator {
45 
+	return func(logCtx Info) (logger Logger, err error) {
46 
+		defer func() {
47 
+			if err != nil {
48 
+				pluginGetter.Get(name, extName, getter.Release)
49 
+			}
50 
+		}()
51 
+		root := filepath.Join(basePath, "run", "docker", "logging")
52 
+		if err := os.MkdirAll(root, 0700); err != nil {
53 
+			return nil, err
54 
+		}
55 
+
56 
+		id := stringid.GenerateNonCryptoID()
57 
+		a := &pluginAdapter{
58 
+			driverName: name,
59 
+			id:         id,
60 
+			plugin:     l,
61 
+			fifoPath:   filepath.Join(root, id),
62 
+			logInfo:    logCtx,
63 
+		}
64 
+
65 
+		cap, err := a.plugin.Capabilities()
66 
+		if err == nil {
67 
+			a.capabilities = cap
68 
+		}
69 
+
70 
+		stream, err := openPluginStream(a)
71 
+		if err != nil {
72 
+			return nil, err
73 
+		}
74 
+
75 
+		a.stream = stream
76 
+		a.enc = logdriver.NewLogEntryEncoder(a.stream)
77 
+
78 
+		if err := l.StartLogging(strings.TrimPrefix(a.fifoPath, basePath), logCtx); err != nil {
79 
+			return nil, errors.Wrapf(err, "error creating logger")
80 
+		}
81 
+
82 
+		if cap.ReadLogs {
83 
+			return &pluginAdapterWithRead{a}, nil
84 
+		}
85 
+
86 
+		return a, nil
87 
+	}
88 
+}
daemon/logger/plugin_unix.go
History View file @ 28334c1
0 89 
new file mode 100644
... ... 
@@ -0,0 +1,20 @@
0 
+// +build linux solaris freebsd
1 
+
2 
+package logger
3 
+
4 
+import (
5 
+	"context"
6 
+	"io"
7 
+
8 
+	"github.com/pkg/errors"
9 
+	"github.com/tonistiigi/fifo"
10 
+	"golang.org/x/sys/unix"
11 
+)
12 
+
13 
+func openPluginStream(a *pluginAdapter) (io.WriteCloser, error) {
14 
+	f, err := fifo.OpenFifo(context.Background(), a.fifoPath, unix.O_WRONLY|unix.O_CREAT|unix.O_NONBLOCK, 0700)
15 
+	if err != nil {
16 
+		return nil, errors.Wrapf(err, "error creating i/o pipe for log plugin: %s", a.Name())
17 
+	}
18 
+	return f, nil
19 
+}
daemon/logger/plugin_unsupported.go
History View file @ 28334c1
0 20 
new file mode 100644
... ... 
@@ -0,0 +1,12 @@
0 
+// +build !linux,!solaris,!freebsd
1 
+
2 
+package logger
3 
+
4 
+import (
5 
+	"errors"
6 
+	"io"
7 
+)
8 
+
9 
+func openPluginStream(a *pluginAdapter) (io.WriteCloser, error) {
10 
+	return nil, errors.New("log plugin not supported")
11 
+}
daemon/logger/proxy.go
History View file @ 28334c1
0 12 
new file mode 100644
... ... 
@@ -0,0 +1,107 @@
0 
+package logger
1 
+
2 
+import (
3 
+	"errors"
4 
+	"io"
5 
+)
6 
+
7 
+type client interface {
8 
+	Call(string, interface{}, interface{}) error
9 
+	Stream(string, interface{}) (io.ReadCloser, error)
10 
+}
11 
+
12 
+type logPluginProxy struct {
13 
+	client
14 
+}
15 
+
16 
+type logPluginProxyStartLoggingRequest struct {
17 
+	File string
18 
+	Info Info
19 
+}
20 
+
21 
+type logPluginProxyStartLoggingResponse struct {
22 
+	Err string
23 
+}
24 
+
25 
+func (pp *logPluginProxy) StartLogging(file string, info Info) (err error) {
26 
+	var (
27 
+		req logPluginProxyStartLoggingRequest
28 
+		ret logPluginProxyStartLoggingResponse
29 
+	)
30 
+
31 
+	req.File = file
32 
+	req.Info = info
33 
+	if err = pp.Call("LogDriver.StartLogging", req, &ret); err != nil {
34 
+		return
35 
+	}
36 
+
37 
+	if ret.Err != "" {
38 
+		err = errors.New(ret.Err)
39 
+	}
40 
+
41 
+	return
42 
+}
43 
+
44 
+type logPluginProxyStopLoggingRequest struct {
45 
+	File string
46 
+}
47 
+
48 
+type logPluginProxyStopLoggingResponse struct {
49 
+	Err string
50 
+}
51 
+
52 
+func (pp *logPluginProxy) StopLogging(file string) (err error) {
53 
+	var (
54 
+		req logPluginProxyStopLoggingRequest
55 
+		ret logPluginProxyStopLoggingResponse
56 
+	)
57 
+
58 
+	req.File = file
59 
+	if err = pp.Call("LogDriver.StopLogging", req, &ret); err != nil {
60 
+		return
61 
+	}
62 
+
63 
+	if ret.Err != "" {
64 
+		err = errors.New(ret.Err)
65 
+	}
66 
+
67 
+	return
68 
+}
69 
+
70 
+type logPluginProxyCapabilitiesResponse struct {
71 
+	Cap Capability
72 
+	Err string
73 
+}
74 
+
75 
+func (pp *logPluginProxy) Capabilities() (cap Capability, err error) {
76 
+	var (
77 
+		ret logPluginProxyCapabilitiesResponse
78 
+	)
79 
+
80 
+	if err = pp.Call("LogDriver.Capabilities", nil, &ret); err != nil {
81 
+		return
82 
+	}
83 
+
84 
+	cap = ret.Cap
85 
+
86 
+	if ret.Err != "" {
87 
+		err = errors.New(ret.Err)
88 
+	}
89 
+
90 
+	return
91 
+}
92 
+
93 
+type logPluginProxyReadLogsRequest struct {
94 
+	Info   Info
95 
+	Config ReadConfig
96 
+}
97 
+
98 
+func (pp *logPluginProxy) ReadLogs(info Info, config ReadConfig) (stream io.ReadCloser, err error) {
99 
+	var (
100 
+		req logPluginProxyReadLogsRequest
101 
+	)
102 
+
103 
+	req.Info = info
104 
+	req.Config = config
105 
+	return pp.Stream("LogDriver.ReadLogs", req)
106 
+}
docs/extend/config.md
History View file @ 28334c1
... ... 
@@ -59,6 +59,8 @@ Config provides the base accessible fields for working with V0 plugin format
59 59
60 60 
         - **docker.authz/1.0**
61 61
62 
+        - **docker.logdriver/1.0**
63 
+
62 64 
     - **`socket`** *string*
63 65
64 66 
       socket is the name of the socket the engine should use to communicate with the plugins.
docs/extend/plugins_logging.md
History View file @ 28334c1
65 67 
new file mode 100644
... ... 
@@ -0,0 +1,220 @@
0 
+---
1 
+title: "Docker log driver plugins"
2 
+description: "Log driver plugins."
3 
+keywords: "Examples, Usage, plugins, docker, documentation, user guide, logging"
4 
+---
5 
+
6 
+<!-- This file is maintained within the docker/docker Github
7 
+     repository at https://github.com/docker/docker/. Make all
8 
+     pull requests against that repo. If you see this file in
9 
+     another repository, consider it read-only there, as it will
10 
+     periodically be overwritten by the definitive file. Pull
11 
+     requests which include edits to this file in other repositories
12 
+     will be rejected.
13 
+-->
14 
+
15 
+# Logging driver plugins
16 
+
17 
+This document describes logging driver plugins for Docker.
18 
+
19 
+Logging drivers enables users to forward container logs to another service for
20 
+processing. Docker includes several logging drivers as built-ins, however can
21 
+never hope to support all use-cases with built-in drivers. Plugins allow Docker
22 
+to support a wide range of logging services without requiring to embed client
23 
+libraries for these services in the main Docker codebase. See the
24 
+[plugin documentation](legacy_plugins.md) for more information.
25 
+
26 
+## Create a logging plugin
27 
+
28 
+The main interface for logging plugins uses the same JSON+HTTP RPC protocol used
29 
+by other plugin types. See the
30 
+[example](https://github.com/cpuguy83/docker-log-driver-test) plugin for a
31 
+reference implementation of a logging plugin. The example wraps the built-in
32 
+`jsonfilelog` log driver.
33 
+
34 
+## LogDriver protocol
35 
+
36 
+Logging plugins must register as a `LogDriver` during plugin activation. Once
37 
+activated users can specify the plugin as a log driver.
38 
+
39 
+There are two HTTP endpoints that logging plugins must implement:
40 
+
41 
+### `/LogDriver.StartLogging`
42 
+
43 
+Signals to the plugin that a container is starting that the plugin should start
44 
+receiving logs for.
45 
+
46 
+Logs will be streamed over the defined file in the request. On Linux this file
47 
+is a FIFO. Logging plugins are not currently supported on Windows.
48 
+
49 
+**Request**:
50 
+```json
51 
+{
52 
+		"File": "/path/to/file/stream",
53 
+		"Info": {
54 
+			"ContainerID": "123456"
55 
+		}
56 
+}
57 
+```
58 
+
59 
+`File` is the path to the log stream that needs to be consumed. Each call to
60 
+`StartLogging` should provide a different file path, even if it's a container
61 
+that the plugin has already received logs for prior. The file is created by
62 
+docker with a randomly generated name.
63 
+
64 
+`Info` is details about the container that's being logged. This is fairly
65 
+free-form, but is defined by the following struct definition:
66 
+
67 
+```go
68 
+type Info struct {
69 
+	Config              map[string]string
70 
+	ContainerID         string
71 
+	ContainerName       string
72 
+	ContainerEntrypoint string
73 
+	ContainerArgs       []string
74 
+	ContainerImageID    string
75 
+	ContainerImageName  string
76 
+	ContainerCreated    time.Time
77 
+	ContainerEnv        []string
78 
+	ContainerLabels     map[string]string
79 
+	LogPath             string
80 
+	DaemonName          string
81 
+}
82 
+```
83 
+
84 
+
85 
+`ContainerID` will always be supplied with this struct, but other fields may be
86 
+empty or missing.
87 
+
88 
+**Response**
89 
+```json
90 
+{
91 
+	"Err": ""
92 
+}
93 
+```
94 
+
95 
+If an error occurred during this request, add an error message to the `Err` field
96 
+in the response. If no error then you can either send an empty response (`{}`)
97 
+or an empty value for the `Err` field.
98 
+
99 
+The driver should at this point be consuming log messages from the passed in file.
100 
+If messages are unconsumed, it may cause the contaier to block while trying to
101 
+write to its stdio streams.
102 
+
103 
+Log stream messages are encoded as protocol buffers. The protobuf definitions are
104 
+in the
105 
+[docker repository](https://github.com/docker/docker/blob/master/api/types/plugins/logdriver/entry.proto).
106 
+
107 
+Since protocol buffers are not self-delimited you must decode them from the stream
108 
+using the following stream format:
109 
+
110 
+```
111 
+[size][message]
112 
+```
113 
+
114 
+Where `size` is a 4-byte big endian binary encoded uint32. `size` in this case
115 
+defines the size of the next message. `message` is the actual log entry.
116 
+
117 
+A reference golang implementation of a stream encoder/decoder can be found
118 
+[here](https://github.com/docker/docker/blob/master/api/types/plugins/logdriver/io.go)
119 
+
120 
+### `/LogDriver.StopLogging`
121 
+
122 
+Signals to the plugin to stop collecting logs from the defined file.
123 
+Once a response is received, the file will be removed by Docker. You must make
124 
+sure to collect all logs on the stream before responding to this request or risk
125 
+losing log data.
126 
+
127 
+Requests on this endpoint does not mean that the container has been removed
128 
+only that it has stopped.
129 
+
130 
+**Request**:
131 
+```json
132 
+{
133 
+		"File": "/path/to/file/stream"
134 
+}
135 
+```
136 
+
137 
+**Response**:
138 
+```json
139 
+{
140 
+	"Err": ""
141 
+}
142 
+```
143 
+
144 
+If an error occurred during this request, add an error message to the `Err` field
145 
+in the response. If no error then you can either send an empty response (`{}`)
146 
+or an empty value for the `Err` field.
147 
+
148 
+## Optional endpoints
149 
+
150 
+Logging plugins can implement two extra logging endpoints:
151 
+
152 
+### `/LogDriver.Capabilities`
153 
+
154 
+Defines the capabilities of the log driver. You must implement this endpoint for
155 
+Docker to be able to take advantage of any of the defined capabilities.
156 
+
157 
+**Request**:
158 
+```json
159 
+{}
160 
+```
161 
+
162 
+**Response**:
163 
+```json
164 
+{
165 
+	"ReadLogs": true
166 
+}
167 
+```
168 
+
169 
+Supported capabilities:
170 
+
171 
+- `ReadLogs` - this tells Docker that the plugin is capable of reading back logs
172 
+to clients. Plugins that report that they support `ReadLogs` must implement the
173 
+`/LogDriver.ReadLogs` endpoint
174 
+
175 
+### `/LogDriver.ReadLogs`
176 
+
177 
+Reads back logs to the client. This is used when `docker logs <container>` is
178 
+called.
179 
+
180 
+In order for Docker to use this endpoint, the plugin must specify as much when
181 
+`/LogDriver.Capabilities` is called.
182 
+
183 
+
184 
+**Request**:
185 
+```json
186 
+{
187 
+	"ReadConfig": {},
188 
+	"Info": {
189 
+		"ContainerID": "123456"
190 
+	}
191 
+}
192 
+```
193 
+
194 
+`ReadConfig` is the list of options for reading, it is defined with the following
195 
+golang struct:
196 
+
197 
+```go
198 
+type ReadConfig struct {
199 
+	Since  time.Time
200 
+	Tail   int
201 
+	Follow bool
202 
+}
203 
+```
204 
+
205 
+- `Since` defines the oldest log that should be sent.
206 
+- `Tail` defines the number of lines to read (e.g. like the command `tail -n 10`)
207 
+- `Follow` signals that the client wants to stay attached to receive new log messages
208 
+as they come in once the existing logs have been read.
209 
+
210 
+`Info` is the same type defined in `/LogDriver.StartLogging`. It should be used
211 
+to determine what set of logs to read.
212 
+
213 
+**Response**:
214 
+```
215 
+{{ log stream }}
216 
+```
217 
+
218 
+The response should be the encoded log message using the same format as the
219 
+messages that the plugin consumed from Docker.
hack/validate/lint
History View file @ 28334c1
... ... 
@@ -4,7 +4,7 @@ export SCRIPTDIR="$( cd "$( dirname "${BASH_SOURCE[0]}" )" && pwd )"
4 4 
 source "${SCRIPTDIR}/.validate"
5 5
6 6 
 IFS=$'\n'
7 
-files=( $(validate_diff --diff-filter=ACMR --name-only -- '*.go' | grep -v '^vendor/' | grep -v '^api/types/container/' | grep -v '^cli/compose/schema/bindata.go' || true) )
7 
+files=( $(validate_diff --diff-filter=ACMR --name-only -- '*.go' | grep -v '^vendor/' | grep -v '^api/types/container/' | grep -v '^cli/compose/schema/bindata.go' | grep -v '^api/types/plugins/logdriver/entry.pb.go' || true) )
8 8 
 unset IFS
9 9
10 10 
 errors=()
integration-cli/docker_cli_plugins_logdriver_test.go
History View file @ 28334c1
11 11 
new file mode 100644
... ... 
@@ -0,0 +1,27 @@
0 
+package main
1 
+
2 
+import (
3 
+	"strings"
4 
+
5 
+	"github.com/docker/docker/integration-cli/checker"
6 
+	"github.com/go-check/check"
7 
+)
8 
+
9 
+func (s *DockerSuite) TestPluginLogDriver(c *check.C) {
10 
+	testRequires(c, IsAmd64, DaemonIsLinux)
11 
+
12 
+	pluginName := "cpuguy83/docker-logdriver-test:latest"
13 
+
14 
+	dockerCmd(c, "plugin", "install", pluginName)
15 
+	dockerCmd(c, "run", "--log-driver", pluginName, "--name=test", "busybox", "echo", "hello")
16 
+	out, _ := dockerCmd(c, "logs", "test")
17 
+	c.Assert(strings.TrimSpace(out), checker.Equals, "hello")
18 
+
19 
+	dockerCmd(c, "start", "-a", "test")
20 
+	out, _ = dockerCmd(c, "logs", "test")
21 
+	c.Assert(strings.TrimSpace(out), checker.Equals, "hello\nhello")
22 
+
23 
+	dockerCmd(c, "rm", "test")
24 
+	dockerCmd(c, "plugin", "disable", pluginName)
25 
+	dockerCmd(c, "plugin", "rm", pluginName)
26 
+}
vendor/github.com/gogo/protobuf/io/full.go
History View file @ 28334c1
0 27 
new file mode 100644
... ... 
@@ -0,0 +1,102 @@
0 
+// Protocol Buffers for Go with Gadgets
1 
+//
2 
+// Copyright (c) 2013, The GoGo Authors. All rights reserved.
3 
+// http://github.com/gogo/protobuf
4 
+//
5 
+// Redistribution and use in source and binary forms, with or without
6 
+// modification, are permitted provided that the following conditions are
7 
+// met:
8 
+//
9 
+//     * Redistributions of source code must retain the above copyright
10 
+// notice, this list of conditions and the following disclaimer.
11 
+//     * Redistributions in binary form must reproduce the above
12 
+// copyright notice, this list of conditions and the following disclaimer
13 
+// in the documentation and/or other materials provided with the
14 
+// distribution.
15 
+//
16 
+// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
17 
+// "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
18 
+// LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
19 
+// A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
20 
+// OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
21 
+// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
22 
+// LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
23 
+// DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
24 
+// THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
25 
+// (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
26 
+// OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
27 
+
28 
+package io
29 
+
30 
+import (
31 
+	"github.com/gogo/protobuf/proto"
32 
+	"io"
33 
+)
34 
+
35 
+func NewFullWriter(w io.Writer) WriteCloser {
36 
+	return &fullWriter{w, nil}
37 
+}
38 
+
39 
+type fullWriter struct {
40 
+	w      io.Writer
41 
+	buffer []byte
42 
+}
43 
+
44 
+func (this *fullWriter) WriteMsg(msg proto.Message) (err error) {
45 
+	var data []byte
46 
+	if m, ok := msg.(marshaler); ok {
47 
+		n, ok := getSize(m)
48 
+		if !ok {
49 
+			data, err = proto.Marshal(msg)
50 
+			if err != nil {
51 
+				return err
52 
+			}
53 
+		}
54 
+		if n >= len(this.buffer) {
55 
+			this.buffer = make([]byte, n)
56 
+		}
57 
+		_, err = m.MarshalTo(this.buffer)
58 
+		if err != nil {
59 
+			return err
60 
+		}
61 
+		data = this.buffer[:n]
62 
+	} else {
63 
+		data, err = proto.Marshal(msg)
64 
+		if err != nil {
65 
+			return err
66 
+		}
67 
+	}
68 
+	_, err = this.w.Write(data)
69 
+	return err
70 
+}
71 
+
72 
+func (this *fullWriter) Close() error {
73 
+	if closer, ok := this.w.(io.Closer); ok {
74 
+		return closer.Close()
75 
+	}
76 
+	return nil
77 
+}
78 
+
79 
+type fullReader struct {
80 
+	r   io.Reader
81 
+	buf []byte
82 
+}
83 
+
84 
+func NewFullReader(r io.Reader, maxSize int) ReadCloser {
85 
+	return &fullReader{r, make([]byte, maxSize)}
86 
+}
87 
+
88 
+func (this *fullReader) ReadMsg(msg proto.Message) error {
89 
+	length, err := this.r.Read(this.buf)
90 
+	if err != nil {
91 
+		return err
92 
+	}
93 
+	return proto.Unmarshal(this.buf[:length], msg)
94 
+}
95 
+
96 
+func (this *fullReader) Close() error {
97 
+	if closer, ok := this.r.(io.Closer); ok {
98 
+		return closer.Close()
99 
+	}
100 
+	return nil
101 
+}
vendor/github.com/gogo/protobuf/io/io.go
History View file @ 28334c1
0 102 
new file mode 100644
... ... 
@@ -0,0 +1,70 @@
0 
+// Protocol Buffers for Go with Gadgets
1 
+//
2 
+// Copyright (c) 2013, The GoGo Authors. All rights reserved.
3 
+// http://github.com/gogo/protobuf
4 
+//
5 
+// Redistribution and use in source and binary forms, with or without
6 
+// modification, are permitted provided that the following conditions are
7 
+// met:
8 
+//
9 
+//     * Redistributions of source code must retain the above copyright
10 
+// notice, this list of conditions and the following disclaimer.
11 
+//     * Redistributions in binary form must reproduce the above
12 
+// copyright notice, this list of conditions and the following disclaimer
13 
+// in the documentation and/or other materials provided with the
14 
+// distribution.
15 
+//
16 
+// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
17 
+// "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
18 
+// LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
19 
+// A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
20 
+// OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
21 
+// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
22 
+// LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
23 
+// DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
24 
+// THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
25 
+// (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
26 
+// OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
27 
+
28 
+package io
29 
+
30 
+import (
31 
+	"github.com/gogo/protobuf/proto"
32 
+	"io"
33 
+)
34 
+
35 
+type Writer interface {
36 
+	WriteMsg(proto.Message) error
37 
+}
38 
+
39 
+type WriteCloser interface {
40 
+	Writer
41 
+	io.Closer
42 
+}
43 
+
44 
+type Reader interface {
45 
+	ReadMsg(msg proto.Message) error
46 
+}
47 
+
48 
+type ReadCloser interface {
49 
+	Reader
50 
+	io.Closer
51 
+}
52 
+
53 
+type marshaler interface {
54 
+	MarshalTo(data []byte) (n int, err error)
55 
+}
56 
+
57 
+func getSize(v interface{}) (int, bool) {
58 
+	if sz, ok := v.(interface {
59 
+		Size() (n int)
60 
+	}); ok {
61 
+		return sz.Size(), true
62 
+	} else if sz, ok := v.(interface {
63 
+		ProtoSize() (n int)
64 
+	}); ok {
65 
+		return sz.ProtoSize(), true
66 
+	} else {
67 
+		return 0, false
68 
+	}
69 
+}
vendor/github.com/gogo/protobuf/io/uint32.go
History View file @ 28334c1
0 70 
new file mode 100644
... ... 
@@ -0,0 +1,126 @@
0 
+// Protocol Buffers for Go with Gadgets
1 
+//
2 
+// Copyright (c) 2013, The GoGo Authors. All rights reserved.
3 
+// http://github.com/gogo/protobuf
4 
+//
5 
+// Redistribution and use in source and binary forms, with or without
6 
+// modification, are permitted provided that the following conditions are
7 
+// met:
8 
+//
9 
+//     * Redistributions of source code must retain the above copyright
10 
+// notice, this list of conditions and the following disclaimer.
11 
+//     * Redistributions in binary form must reproduce the above
12 
+// copyright notice, this list of conditions and the following disclaimer
13 
+// in the documentation and/or other materials provided with the
14 
+// distribution.
15 
+//
16 
+// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
17 
+// "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
18 
+// LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
19 
+// A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
20 
+// OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
21 
+// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
22 
+// LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
23 
+// DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
24 
+// THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
25 
+// (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
26 
+// OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
27 
+
28 
+package io
29 
+
30 
+import (
31 
+	"encoding/binary"
32 
+	"github.com/gogo/protobuf/proto"
33 
+	"io"
34 
+)
35 
+
36 
+func NewUint32DelimitedWriter(w io.Writer, byteOrder binary.ByteOrder) WriteCloser {
37 
+	return &uint32Writer{w, byteOrder, nil}
38 
+}
39 
+
40 
+func NewSizeUint32DelimitedWriter(w io.Writer, byteOrder binary.ByteOrder, size int) WriteCloser {
41 
+	return &uint32Writer{w, byteOrder, make([]byte, size)}
42 
+}
43 
+
44 
+type uint32Writer struct {
45 
+	w         io.Writer
46 
+	byteOrder binary.ByteOrder
47 
+	buffer    []byte
48 
+}
49 
+
50 
+func (this *uint32Writer) WriteMsg(msg proto.Message) (err error) {
51 
+	var data []byte
52 
+	if m, ok := msg.(marshaler); ok {
53 
+		n, ok := getSize(m)
54 
+		if !ok {
55 
+			data, err = proto.Marshal(msg)
56 
+			if err != nil {
57 
+				return err
58 
+			}
59 
+		}
60 
+		if n >= len(this.buffer) {
61 
+			this.buffer = make([]byte, n)
62 
+		}
63 
+		_, err = m.MarshalTo(this.buffer)
64 
+		if err != nil {
65 
+			return err
66 
+		}
67 
+		data = this.buffer[:n]
68 
+	} else {
69 
+		data, err = proto.Marshal(msg)
70 
+		if err != nil {
71 
+			return err
72 
+		}
73 
+	}
74 
+	length := uint32(len(data))
75 
+	if err = binary.Write(this.w, this.byteOrder, &length); err != nil {
76 
+		return err
77 
+	}
78 
+	_, err = this.w.Write(data)
79 
+	return err
80 
+}
81 
+
82 
+func (this *uint32Writer) Close() error {
83 
+	if closer, ok := this.w.(io.Closer); ok {
84 
+		return closer.Close()
85 
+	}
86 
+	return nil
87 
+}
88 
+
89 
+type uint32Reader struct {
90 
+	r         io.Reader
91 
+	byteOrder binary.ByteOrder
92 
+	lenBuf    []byte
93 
+	buf       []byte
94 
+	maxSize   int
95 
+}
96 
+
97 
+func NewUint32DelimitedReader(r io.Reader, byteOrder binary.ByteOrder, maxSize int) ReadCloser {
98 
+	return &uint32Reader{r, byteOrder, make([]byte, 4), nil, maxSize}
99 
+}
100 
+
101 
+func (this *uint32Reader) ReadMsg(msg proto.Message) error {
102 
+	if _, err := io.ReadFull(this.r, this.lenBuf); err != nil {
103 
+		return err
104 
+	}
105 
+	length32 := this.byteOrder.Uint32(this.lenBuf)
106 
+	length := int(length32)
107 
+	if length < 0 || length > this.maxSize {
108 
+		return io.ErrShortBuffer
109 
+	}
110 
+	if length >= len(this.buf) {
111 
+		this.buf = make([]byte, length)
112 
+	}
113 
+	_, err := io.ReadFull(this.r, this.buf[:length])
114 
+	if err != nil {
115 
+		return err
116 
+	}
117 
+	return proto.Unmarshal(this.buf[:length], msg)
118 
+}
119 
+
120 
+func (this *uint32Reader) Close() error {
121 
+	if closer, ok := this.r.(io.Closer); ok {
122 
+		return closer.Close()
123 
+	}
124 
+	return nil
125 
+}
vendor/github.com/gogo/protobuf/io/varint.go
History View file @ 28334c1
0 126 
new file mode 100644
... ... 
@@ -0,0 +1,134 @@
0 
+// Protocol Buffers for Go with Gadgets
1 
+//
2 
+// Copyright (c) 2013, The GoGo Authors. All rights reserved.
3 
+// http://github.com/gogo/protobuf
4 
+//
5 
+// Redistribution and use in source and binary forms, with or without
6 
+// modification, are permitted provided that the following conditions are
7 
+// met:
8 
+//
9 
+//     * Redistributions of source code must retain the above copyright
10 
+// notice, this list of conditions and the following disclaimer.
11 
+//     * Redistributions in binary form must reproduce the above
12 
+// copyright notice, this list of conditions and the following disclaimer
13 
+// in the documentation and/or other materials provided with the
14 
+// distribution.
15 
+//
16 
+// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
17 
+// "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
18 
+// LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
19 
+// A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
20 
+// OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
21 
+// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
22 
+// LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
23 
+// DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
24 
+// THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
25 
+// (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
26 
+// OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
27 
+
28 
+package io
29 
+
30 
+import (
31 
+	"bufio"
32 
+	"encoding/binary"
33 
+	"errors"
34 
+	"github.com/gogo/protobuf/proto"
35 
+	"io"
36 
+)
37 
+
38 
+var (
39 
+	errSmallBuffer = errors.New("Buffer Too Small")
40 
+	errLargeValue  = errors.New("Value is Larger than 64 bits")
41 
+)
42 
+
43 
+func NewDelimitedWriter(w io.Writer) WriteCloser {
44 
+	return &varintWriter{w, make([]byte, 10), nil}
45 
+}
46 
+
47 
+type varintWriter struct {
48 
+	w      io.Writer
49 
+	lenBuf []byte
50 
+	buffer []byte
51 
+}
52 
+
53 
+func (this *varintWriter) WriteMsg(msg proto.Message) (err error) {
54 
+	var data []byte
55 
+	if m, ok := msg.(marshaler); ok {
56 
+		n, ok := getSize(m)
57 
+		if !ok {
58 
+			data, err = proto.Marshal(msg)
59 
+			if err != nil {
60 
+				return err
61 
+			}
62 
+		}
63 
+		if n >= len(this.buffer) {
64 
+			this.buffer = make([]byte, n)
65 
+		}
66 
+		_, err = m.MarshalTo(this.buffer)
67 
+		if err != nil {
68 
+			return err
69 
+		}
70 
+		data = this.buffer[:n]
71 
+	} else {
72 
+		data, err = proto.Marshal(msg)
73 
+		if err != nil {
74 
+			return err
75 
+		}
76 
+	}
77 
+	length := uint64(len(data))
78 
+	n := binary.PutUvarint(this.lenBuf, length)
79 
+	_, err = this.w.Write(this.lenBuf[:n])
80 
+	if err != nil {
81 
+		return err
82 
+	}
83 
+	_, err = this.w.Write(data)
84 
+	return err
85 
+}
86 
+
87 
+func (this *varintWriter) Close() error {
88 
+	if closer, ok := this.w.(io.Closer); ok {
89 
+		return closer.Close()
90 
+	}
91 
+	return nil
92 
+}
93 
+
94 
+func NewDelimitedReader(r io.Reader, maxSize int) ReadCloser {
95 
+	var closer io.Closer
96 
+	if c, ok := r.(io.Closer); ok {
97 
+		closer = c
98 
+	}
99 
+	return &varintReader{bufio.NewReader(r), nil, maxSize, closer}
100 
+}
101 
+
102 
+type varintReader struct {
103 
+	r       *bufio.Reader
104 
+	buf     []byte
105 
+	maxSize int
106 
+	closer  io.Closer
107 
+}
108 
+
109 
+func (this *varintReader) ReadMsg(msg proto.Message) error {
110 
+	length64, err := binary.ReadUvarint(this.r)
111 
+	if err != nil {
112 
+		return err
113 
+	}
114 
+	length := int(length64)
115 
+	if length < 0 || length > this.maxSize {
116 
+		return io.ErrShortBuffer
117 
+	}
118 
+	if len(this.buf) < length {
119 
+		this.buf = make([]byte, length)
120 
+	}
121 
+	buf := this.buf[:length]
122 
+	if _, err := io.ReadFull(this.r, buf); err != nil {
123 
+		return err
124 
+	}
125 
+	return proto.Unmarshal(buf, msg)
126 
+}
127 
+
128 
+func (this *varintReader) Close() error {
129 
+	if this.closer != nil {
130 
+		return this.closer.Close()
131 
+	}
132 
+	return nil
133 
+}