Merge component 'engine' from git@github.com:moby/moby master

components/engine/api/types/backend/backend.go

@@ -25,17 +25,27 @@ type ContainerAttachConfig struct {
 	MuxStreams bool
 }
 
+// PartialLogMetaData provides meta data for a partial log message. Messages
+// exceeding a predefined size are split into chunks with this metadata. The
+// expectation is for the logger endpoints to assemble the chunks using this
+// metadata.
+type PartialLogMetaData struct {
+	Last    bool   //true if this message is last of a partial
+	ID      string // identifies group of messages comprising a single record
+	Ordinal int    // ordering of message in partial group
+}
+
 // LogMessage is datastructure that represents piece of output produced by some
 // container.  The Line member is a slice of an array whose contents can be
 // changed after a log driver's Log() method returns.
 // changes to this struct need to be reflect in the reset method in
 // daemon/logger/logger.go
 type LogMessage struct {
-	Line      []byte
-	Source    string
-	Timestamp time.Time
-	Attrs     []LogAttr
-	Partial   bool
+	Line         []byte
+	Source       string
+	Timestamp    time.Time
+	Attrs        []LogAttr
+	PLogMetaData *PartialLogMetaData
 
 	// Err is an error associated with a message. Completeness of a message
 	// with Err is not expected, tho it may be partially complete (fields may

components/engine/daemon/logger/adapter.go

@@ -37,7 +37,7 @@ func (a *pluginAdapter) Log(msg *Message) error {
 
 	a.buf.Line = msg.Line
 	a.buf.TimeNano = msg.Timestamp.UnixNano()
-	a.buf.Partial = msg.Partial
+	a.buf.Partial = (msg.PLogMetaData != nil)
 	a.buf.Source = msg.Source
 
 	err := a.enc.Encode(&a.buf)

components/engine/daemon/logger/copier.go

@@ -6,6 +6,8 @@ import (
 	"sync"
 	"time"
 
+	types "github.com/docker/docker/api/types/backend"
+	"github.com/docker/docker/pkg/stringid"
 	"github.com/sirupsen/logrus"
 )
 
@@ -58,6 +60,11 @@ func (c *Copier) copySrc(name string, src io.Reader) {
 
 	n := 0
 	eof := false
+	var partialid string
+	var partialTS time.Time
+	var ordinal int
+	firstPartial := true
+	hasMorePartial := false
 
 	for {
 		select {
@@ -87,6 +94,7 @@ func (c *Copier) copySrc(name string, src io.Reader) {
 			}
 			// Break up the data that we've buffered up into lines, and log each in turn.
 			p := 0
+
 			for q := bytes.IndexByte(buf[p:n], '\n'); q >= 0; q = bytes.IndexByte(buf[p:n], '\n') {
 				select {
 				case <-c.closed:
@@ -94,9 +102,23 @@ func (c *Copier) copySrc(name string, src io.Reader) {
 				default:
 					msg := NewMessage()
 					msg.Source = name
-					msg.Timestamp = time.Now().UTC()
 					msg.Line = append(msg.Line, buf[p:p+q]...)
 
+					if hasMorePartial {
+						msg.PLogMetaData = &types.PartialLogMetaData{ID: partialid, Ordinal: ordinal, Last: true}
+
+						// reset
+						partialid = ""
+						ordinal = 0
+						firstPartial = true
+						hasMorePartial = false
+					}
+					if msg.PLogMetaData == nil {
+						msg.Timestamp = time.Now().UTC()
+					} else {
+						msg.Timestamp = partialTS
+					}
+
 					if logErr := c.dst.Log(msg); logErr != nil {
 						logrus.Errorf("Failed to log msg %q for logger %s: %s", msg.Line, c.dst.Name(), logErr)
 					}
@@ -110,9 +132,23 @@ func (c *Copier) copySrc(name string, src io.Reader) {
 				if p < n {
 					msg := NewMessage()
 					msg.Source = name
-					msg.Timestamp = time.Now().UTC()
 					msg.Line = append(msg.Line, buf[p:n]...)
-					msg.Partial = true
+
+					// Generate unique partialID for first partial. Use it across partials.
+					// Record timestamp for first partial. Use it across partials.
+					// Initialize Ordinal for first partial. Increment it across partials.
+					if firstPartial {
+						msg.Timestamp = time.Now().UTC()
+						partialTS = msg.Timestamp
+						partialid = stringid.GenerateRandomID()
+						ordinal = 1
+						firstPartial = false
+					} else {
+						msg.Timestamp = partialTS
+					}
+					msg.PLogMetaData = &types.PartialLogMetaData{ID: partialid, Ordinal: ordinal, Last: false}
+					ordinal++
+					hasMorePartial = true
 
 					if logErr := c.dst.Log(msg); logErr != nil {
 						logrus.Errorf("Failed to log msg %q for logger %s: %s", msg.Line, c.dst.Name(), logErr)

components/engine/daemon/logger/copier_test.go

@@ -258,6 +258,141 @@ func TestCopierWithSized(t *testing.T) {
 	}
 }
 
+func checkIdentical(t *testing.T, msg Message, expectedID string, expectedTS time.Time) {
+	if msg.PLogMetaData.ID != expectedID {
+		t.Fatalf("IDs are not he same across partials. Expected: %s Received: %s",
+			expectedID, msg.PLogMetaData.ID)
+	}
+	if msg.Timestamp != expectedTS {
+		t.Fatalf("Timestamps are not the same across partials. Expected: %v Received: %v",
+			expectedTS.Format(time.UnixDate), msg.Timestamp.Format(time.UnixDate))
+	}
+}
+
+// Have long lines and make sure that it comes out with PartialMetaData
+func TestCopierWithPartial(t *testing.T) {
+	stdoutLongLine := strings.Repeat("a", defaultBufSize)
+	stderrLongLine := strings.Repeat("b", defaultBufSize)
+	stdoutTrailingLine := "stdout trailing line"
+	stderrTrailingLine := "stderr trailing line"
+	normalStr := "This is an impartial message :)"
+
+	var stdout bytes.Buffer
+	var stderr bytes.Buffer
+	var normalMsg bytes.Buffer
+
+	for i := 0; i < 3; i++ {
+		if _, err := stdout.WriteString(stdoutLongLine); err != nil {
+			t.Fatal(err)
+		}
+		if _, err := stderr.WriteString(stderrLongLine); err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	if _, err := stdout.WriteString(stdoutTrailingLine + "\n"); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := stderr.WriteString(stderrTrailingLine + "\n"); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := normalMsg.WriteString(normalStr + "\n"); err != nil {
+		t.Fatal(err)
+	}
+
+	var jsonBuf bytes.Buffer
+
+	jsonLog := &TestLoggerJSON{Encoder: json.NewEncoder(&jsonBuf)}
+
+	c := NewCopier(
+		map[string]io.Reader{
+			"stdout": &stdout,
+			"normal": &normalMsg,
+			"stderr": &stderr,
+		},
+		jsonLog)
+	c.Run()
+	wait := make(chan struct{})
+	go func() {
+		c.Wait()
+		close(wait)
+	}()
+	select {
+	case <-time.After(1 * time.Second):
+		t.Fatal("Copier failed to do its work in 1 second")
+	case <-wait:
+	}
+
+	dec := json.NewDecoder(&jsonBuf)
+	expectedMsgs := 9
+	recvMsgs := 0
+	var expectedPartID1, expectedPartID2 string
+	var expectedTS1, expectedTS2 time.Time
+
+	for {
+		var msg Message
+
+		if err := dec.Decode(&msg); err != nil {
+			if err == io.EOF {
+				break
+			}
+			t.Fatal(err)
+		}
+		if msg.Source != "stdout" && msg.Source != "stderr" && msg.Source != "normal" {
+			t.Fatalf("Wrong Source: %q, should be %q or %q or %q", msg.Source, "stdout", "stderr", "normal")
+		}
+
+		if msg.Source == "stdout" {
+			if string(msg.Line) != stdoutLongLine && string(msg.Line) != stdoutTrailingLine {
+				t.Fatalf("Wrong Line: %q, expected 'stdoutLongLine' or 'stdoutTrailingLine'", msg.Line)
+			}
+
+			if msg.PLogMetaData.ID == "" {
+				t.Fatalf("Expected partial metadata. Got nothing")
+			}
+
+			if msg.PLogMetaData.Ordinal == 1 {
+				expectedPartID1 = msg.PLogMetaData.ID
+				expectedTS1 = msg.Timestamp
+			} else {
+				checkIdentical(t, msg, expectedPartID1, expectedTS1)
+			}
+			if msg.PLogMetaData.Ordinal == 4 && !msg.PLogMetaData.Last {
+				t.Fatalf("Last is not set for last chunk")
+			}
+		}
+
+		if msg.Source == "stderr" {
+			if string(msg.Line) != stderrLongLine && string(msg.Line) != stderrTrailingLine {
+				t.Fatalf("Wrong Line: %q, expected 'stderrLongLine' or 'stderrTrailingLine'", msg.Line)
+			}
+
+			if msg.PLogMetaData.ID == "" {
+				t.Fatalf("Expected partial metadata. Got nothing")
+			}
+
+			if msg.PLogMetaData.Ordinal == 1 {
+				expectedPartID2 = msg.PLogMetaData.ID
+				expectedTS2 = msg.Timestamp
+			} else {
+				checkIdentical(t, msg, expectedPartID2, expectedTS2)
+			}
+			if msg.PLogMetaData.Ordinal == 4 && !msg.PLogMetaData.Last {
+				t.Fatalf("Last is not set for last chunk")
+			}
+		}
+
+		if msg.Source == "normal" && msg.PLogMetaData != nil {
+			t.Fatalf("Normal messages should not have PartialLogMetaData")
+		}
+		recvMsgs++
+	}
+
+	if expectedMsgs != recvMsgs {
+		t.Fatalf("Expected msgs: %d Recv msgs: %d", expectedMsgs, recvMsgs)
+	}
+}
+
 type BenchmarkLoggerDummy struct {
 }
 

components/engine/daemon/logger/journald/journald.go

@@ -108,7 +108,7 @@ func (s *journald) Log(msg *logger.Message) error {
 	for k, v := range s.vars {
 		vars[k] = v
 	}
-	if msg.Partial {
+	if msg.PLogMetaData != nil {
 		vars["CONTAINER_PARTIAL_MESSAGE"] = "true"
 	}
 

components/engine/daemon/logger/jsonfilelog/jsonfilelog.go

@@ -132,7 +132,7 @@ func (l *JSONFileLogger) Log(msg *logger.Message) error {
 
 func marshalMessage(msg *logger.Message, extra json.RawMessage, buf *bytes.Buffer) error {
 	logLine := msg.Line
-	if !msg.Partial {
+	if msg.PLogMetaData == nil || (msg.PLogMetaData != nil && msg.PLogMetaData.Last) {
 		logLine = append(msg.Line, '\n')
 	}
 	err := (&jsonlog.JSONLogs{

components/engine/daemon/logger/logger.go

@@ -60,7 +60,7 @@ func (m *Message) reset() {
 	m.Line = m.Line[:0]
 	m.Source = ""
 	m.Attrs = nil
-	m.Partial = false
+	m.PLogMetaData = nil
 
 	m.Err = nil
 }

components/engine/daemon/logger/logger_test.go

@@ -6,9 +6,9 @@ import (
 
 func (m *Message) copy() *Message {
 	msg := &Message{
-		Source:    m.Source,
-		Partial:   m.Partial,
-		Timestamp: m.Timestamp,
+		Source:       m.Source,
+		PLogMetaData: m.PLogMetaData,
+		Timestamp:    m.Timestamp,
 	}
 
 	if m.Attrs != nil {

components/engine/hack/ci/janky

@@ -6,7 +6,7 @@ hack/validate/default
 hack/test/unit
 bash <(curl -s https://codecov.io/bash) \
     -f coverage.txt \
-    -C $GIT_SHA1 || \
+    -C "$GIT_SHA1" || \
     echo 'Codecov failed to upload'
 
 hack/make.sh \

components/engine/hack/dockerfile/install/proxy.installer

@@ -3,7 +3,7 @@
 # LIBNETWORK_COMMIT is used to build the docker-userland-proxy binary. When
 # updating the binary version, consider updating github.com/docker/libnetwork
 # in vendor.conf accordingly
-LIBNETWORK_COMMIT=5c1218c956c99f3365711974e300087810c31379
+LIBNETWORK_COMMIT=c15b372ef22125880d378167dde44f4b134e1a77
 
 install_proxy() {
 	case "$1" in

components/engine/integration/plugin/authz/authz_plugin_test.go

@@ -23,6 +23,7 @@ import (
 	"github.com/docker/docker/client"
 	"github.com/docker/docker/integration/internal/container"
 	"github.com/docker/docker/internal/test/environment"
+	"github.com/docker/docker/pkg/archive"
 	"github.com/docker/docker/pkg/authorization"
 	"github.com/gotestyourself/gotestyourself/assert"
 	"github.com/gotestyourself/gotestyourself/skip"
@@ -382,6 +383,56 @@ func TestAuthZPluginEnsureLoadImportWorking(t *testing.T) {
 	assert.NilError(t, err)
 }
 
+func TestAuthzPluginEnsureContainerCopyToFrom(t *testing.T) {
+	defer setupTestV1(t)()
+	ctrl.reqRes.Allow = true
+	ctrl.resRes.Allow = true
+	d.StartWithBusybox(t, "--authorization-plugin="+testAuthZPlugin, "--authorization-plugin="+testAuthZPlugin)
+
+	dir, err := ioutil.TempDir("", t.Name())
+	assert.Assert(t, err)
+	defer os.RemoveAll(dir)
+
+	f, err := ioutil.TempFile(dir, "send")
+	assert.Assert(t, err)
+	defer f.Close()
+
+	buf := make([]byte, 1024)
+	fileSize := len(buf) * 1024 * 10
+	for written := 0; written < fileSize; {
+		n, err := f.Write(buf)
+		assert.Assert(t, err)
+		written += n
+	}
+
+	ctx := context.Background()
+	client, err := d.NewClient()
+	assert.Assert(t, err)
+
+	cID := container.Run(t, ctx, client)
+	defer client.ContainerRemove(ctx, cID, types.ContainerRemoveOptions{Force: true})
+
+	_, err = f.Seek(0, io.SeekStart)
+	assert.Assert(t, err)
+
+	srcInfo, err := archive.CopyInfoSourcePath(f.Name(), false)
+	assert.Assert(t, err)
+	srcArchive, err := archive.TarResource(srcInfo)
+	assert.Assert(t, err)
+	defer srcArchive.Close()
+
+	dstDir, preparedArchive, err := archive.PrepareArchiveCopy(srcArchive, srcInfo, archive.CopyInfo{Path: "/test"})
+	assert.Assert(t, err)
+
+	err = client.CopyToContainer(ctx, cID, dstDir, preparedArchive, types.CopyToContainerOptions{})
+	assert.Assert(t, err)
+
+	rdr, _, err := client.CopyFromContainer(ctx, cID, "/test")
+	assert.Assert(t, err)
+	_, err = io.Copy(ioutil.Discard, rdr)
+	assert.Assert(t, err)
+}
+
 func imageSave(client client.APIClient, path, image string) error {
 	ctx := context.Background()
 	responseReader, err := client.ImageSave(ctx, []string{image})

components/engine/pkg/authorization/authz.go

@@ -5,6 +5,7 @@ import (
 	"bytes"
 	"fmt"
 	"io"
+	"mime"
 	"net/http"
 	"strings"
 
@@ -153,7 +154,12 @@ func sendBody(url string, header http.Header) bool {
 	}
 
 	// body is sent only for text or json messages
-	return header.Get("Content-Type") == "application/json"
+	contentType, _, err := mime.ParseMediaType(header.Get("Content-Type"))
+	if err != nil {
+		return false
+	}
+
+	return contentType == "application/json"
 }
 
 // headers returns flatten version of the http headers excluding authorization

components/engine/pkg/authorization/authz_unix_test.go

@@ -172,6 +172,66 @@ func TestDrainBody(t *testing.T) {
 	}
 }
 
+func TestSendBody(t *testing.T) {
+	var (
+		url       = "nothing.com"
+		testcases = []struct {
+			contentType string
+			expected    bool
+		}{
+			{
+				contentType: "application/json",
+				expected:    true,
+			},
+			{
+				contentType: "Application/json",
+				expected:    true,
+			},
+			{
+				contentType: "application/JSON",
+				expected:    true,
+			},
+			{
+				contentType: "APPLICATION/JSON",
+				expected:    true,
+			},
+			{
+				contentType: "application/json; charset=utf-8",
+				expected:    true,
+			},
+			{
+				contentType: "application/json;charset=utf-8",
+				expected:    true,
+			},
+			{
+				contentType: "application/json; charset=UTF8",
+				expected:    true,
+			},
+			{
+				contentType: "application/json;charset=UTF8",
+				expected:    true,
+			},
+			{
+				contentType: "text/html",
+				expected:    false,
+			},
+			{
+				contentType: "",
+				expected:    false,
+			},
+		}
+	)
+
+	for _, testcase := range testcases {
+		header := http.Header{}
+		header.Set("Content-Type", testcase.contentType)
+
+		if b := sendBody(url, header); b != testcase.expected {
+			t.Fatalf("Unexpected Content-Type; Expected: %t, Actual: %t", testcase.expected, b)
+		}
+	}
+}
+
 func TestResponseModifierOverride(t *testing.T) {
 	r := httptest.NewRecorder()
 	m := NewResponseModifier(r)

components/engine/pkg/authorization/response.go

@@ -47,6 +47,8 @@ func NewResponseModifier(rw http.ResponseWriter) ResponseModifier {
 	return &responseModifier{rw: rw, header: make(http.Header)}
 }
 
+const maxBufferSize = 64 * 1024
+
 // responseModifier is used as an adapter to http.ResponseWriter in order to manipulate and explore
 // the http request/response from docker daemon
 type responseModifier struct {
@@ -116,11 +118,13 @@ func (rm *responseModifier) OverrideHeader(b []byte) error {
 
 // Write stores the byte array inside content
 func (rm *responseModifier) Write(b []byte) (int, error) {
-
 	if rm.hijacked {
 		return rm.rw.Write(b)
 	}
 
+	if len(rm.body)+len(b) > maxBufferSize {
+		rm.Flush()
+	}
 	rm.body = append(rm.body, b...)
 	return len(b), nil
 }
@@ -192,11 +196,14 @@ func (rm *responseModifier) FlushAll() error {
 	var err error
 	if len(rm.body) > 0 {
 		// Write body
-		_, err = rm.rw.Write(rm.body)
+		var n int
+		n, err = rm.rw.Write(rm.body)
+		// TODO(@cpuguy83): there is now a relatively small buffer limit, instead of discarding our buffer here and
+		// allocating again later this should just keep using the same buffer and track the buffer position (like a bytes.Buffer with a fixed size)
+		rm.body = rm.body[n:]
 	}
 
 	// Clean previous data
-	rm.body = nil
 	rm.statusCode = 0
 	rm.header = http.Header{}
 	return err

components/engine/vendor.conf

@@ -32,7 +32,7 @@ github.com/tonistiigi/fsutil dea3a0da73aee887fc02142d995be764106ac5e2
 #get libnetwork packages
 
 # When updating, also update LIBNETWORK_COMMIT in hack/dockerfile/install/proxy accordingly
-github.com/docker/libnetwork 5c1218c956c99f3365711974e300087810c31379
+github.com/docker/libnetwork c15b372ef22125880d378167dde44f4b134e1a77
 github.com/docker/go-events 9461782956ad83b30282bf90e31fa6a70c255ba9
 github.com/armon/go-radix e39d623f12e8e41c7b5529e9a9dd67a1e2261f80
 github.com/armon/go-metrics eb0af217e5e9747e41dd5303755356b62d28e3ec

components/engine/vendor/github.com/docker/libnetwork/drivers/overlay/ostweaks_linux.go

@@ -3,15 +3,36 @@ package overlay
 import (
 	"io/ioutil"
 	"path"
+	"strconv"
 	"strings"
 
 	"github.com/sirupsen/logrus"
 )
 
-var sysctlConf = map[string]string{
-	"net.ipv4.neigh.default.gc_thresh1": "8192",
-	"net.ipv4.neigh.default.gc_thresh2": "49152",
-	"net.ipv4.neigh.default.gc_thresh3": "65536",
+type conditionalCheck func(val1, val2 string) bool
+
+type osValue struct {
+	value   string
+	checkFn conditionalCheck
+}
+
+var osConfig = map[string]osValue{
+	"net.ipv4.neigh.default.gc_thresh1": {"8192", checkHigher},
+	"net.ipv4.neigh.default.gc_thresh2": {"49152", checkHigher},
+	"net.ipv4.neigh.default.gc_thresh3": {"65536", checkHigher},
+}
+
+func propertyIsValid(val1, val2 string, check conditionalCheck) bool {
+	if check == nil || check(val1, val2) {
+		return true
+	}
+	return false
+}
+
+func checkHigher(val1, val2 string) bool {
+	val1Int, _ := strconv.ParseInt(val1, 10, 32)
+	val2Int, _ := strconv.ParseInt(val2, 10, 32)
+	return val1Int < val2Int
 }
 
 // writeSystemProperty writes the value to a path under /proc/sys as determined from the key.
@@ -21,10 +42,31 @@ func writeSystemProperty(key, value string) error {
 	return ioutil.WriteFile(path.Join("/proc/sys", keyPath), []byte(value), 0644)
 }
 
+func readSystemProperty(key string) (string, error) {
+	keyPath := strings.Replace(key, ".", "/", -1)
+	value, err := ioutil.ReadFile(path.Join("/proc/sys", keyPath))
+	if err != nil {
+		return "", err
+	}
+	return string(value), nil
+}
+
 func applyOStweaks() {
-	for k, v := range sysctlConf {
-		if err := writeSystemProperty(k, v); err != nil {
-			logrus.Errorf("error setting the kernel parameter %s = %s, err: %s", k, v, err)
+	for k, v := range osConfig {
+		// read the existing property from disk
+		oldv, err := readSystemProperty(k)
+		if err != nil {
+			logrus.Errorf("error reading the kernel parameter %s, error: %s", k, err)
+			continue
+		}
+
+		if propertyIsValid(oldv, v.value, v.checkFn) {
+			// write new prop value to disk
+			if err := writeSystemProperty(k, v.value); err != nil {
+				logrus.Errorf("error setting the kernel parameter %s = %s, (leaving as %s) error: %s", k, v.value, oldv, err)
+				continue
+			}
+			logrus.Debugf("updated kernel parameter %s = %s (was %s)", k, v.value, oldv)
 		}
 	}
 }

components/engine/vendor/github.com/docker/libnetwork/networkdb/delegate.go

@@ -21,8 +21,8 @@ func (nDB *NetworkDB) handleNodeEvent(nEvent *NodeEvent) bool {
 	// time.
 	nDB.networkClock.Witness(nEvent.LTime)
 
-	nDB.RLock()
-	defer nDB.RUnlock()
+	nDB.Lock()
+	defer nDB.Unlock()
 
 	// check if the node exists
 	n, _, _ := nDB.findNode(nEvent.NodeName)