1. docker
  2. Commit 27bd6842f8518780b99fdb42f8e7f84c17856d87
Browse code

Implement plugins for logging drivers

Logging plugins use the same HTTP interface as other plugins for basic
command operations meanwhile actual logging operations are handled (on
Unix) via a fifo.

The plugin interface looks like so:

```go
type loggingPlugin interface {
StartLogging(fifoPath string, loggingContext Context) error
StopLogging(fifoPath)
```

This means a plugin must implement `LoggingDriver.StartLogging` and
`LoggingDriver.StopLogging` endpoints and be able to consume the passed
in fifo.

Logs are sent via stream encoder to the fifo encoded with protobuf.

Signed-off-by: Brian Goff <cpuguy83@gmail.com>

Brian Goff authored on 2016/11/15 03:53:53
Showing 22 changed files
api/types/plugins/logdriver/entry.pb.go
History View file @ 27bd684
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 @ 27bd684
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 @ 27bd684
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 @ 27bd684
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 @ 27bd684
... ... 
@@ -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"
... ... 
@@ -589,6 +590,7 @@ func NewDaemon(config *config.Config, registryService registry.Service, containe
589 589
590 590 
 	d.RegistryService = registryService
591 591 
 	d.PluginStore = pluginStore
592 
+	logger.RegisterPluginGetter(d.PluginStore)
592 593
593 594 
 	// Plugin system initialization should happen before restore. Do not change order.
594 595 
 	d.pluginManager, err = plugin.NewManager(plugin.ManagerConfig{
daemon/logger/adapter.go
History View file @ 27bd684
595 596 
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 @ 27bd684
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 @ 27bd684
... ... 
@@ -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 @ 27bd684
... ... 
@@ -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 @ 27bd684
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 @ 27bd684
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 @ 27bd684
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 @ 27bd684
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 @ 27bd684
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 @ 27bd684
... ... 
@@ -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 @ 27bd684
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 @ 27bd684
... ... 
@@ -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 @ 27bd684
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 @ 27bd684
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 @ 27bd684
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 @ 27bd684
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 @ 27bd684
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 
+}