diff --git a/components/engine/api/types/backend/backend.go b/components/engine/api/types/backend/backend.go index ffd556e944..ef1e669c39 100644 --- a/components/engine/api/types/backend/backend.go +++ b/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 diff --git a/components/engine/daemon/logger/adapter.go b/components/engine/daemon/logger/adapter.go index 71d033f0a9..5b9252d324 100644 --- a/components/engine/daemon/logger/adapter.go +++ b/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) diff --git a/components/engine/daemon/logger/copier.go b/components/engine/daemon/logger/copier.go index e38176dc74..ae86777f33 100644 --- a/components/engine/daemon/logger/copier.go +++ b/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) diff --git a/components/engine/daemon/logger/copier_test.go b/components/engine/daemon/logger/copier_test.go index 62e1a56ecc..d09450bd19 100644 --- a/components/engine/daemon/logger/copier_test.go +++ b/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 { } diff --git a/components/engine/daemon/logger/journald/journald.go b/components/engine/daemon/logger/journald/journald.go index 47dbfb920c..342e18f57f 100644 --- a/components/engine/daemon/logger/journald/journald.go +++ b/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" } diff --git a/components/engine/daemon/logger/jsonfilelog/jsonfilelog.go b/components/engine/daemon/logger/jsonfilelog/jsonfilelog.go index 2b1e91d063..7d0533ec84 100644 --- a/components/engine/daemon/logger/jsonfilelog/jsonfilelog.go +++ b/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{ diff --git a/components/engine/daemon/logger/logger.go b/components/engine/daemon/logger/logger.go index b64d77840e..912e855c7f 100644 --- a/components/engine/daemon/logger/logger.go +++ b/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 } diff --git a/components/engine/daemon/logger/logger_test.go b/components/engine/daemon/logger/logger_test.go index 3e29e22801..eaeec24085 100644 --- a/components/engine/daemon/logger/logger_test.go +++ b/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 { diff --git a/components/engine/hack/ci/janky b/components/engine/hack/ci/janky index c77cd022a4..f2bdfbf326 100755 --- a/components/engine/hack/ci/janky +++ b/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 \ diff --git a/components/engine/hack/dockerfile/install/proxy.installer b/components/engine/hack/dockerfile/install/proxy.installer index d66a5ffe40..dcc190bbbe 100755 --- a/components/engine/hack/dockerfile/install/proxy.installer +++ b/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 diff --git a/components/engine/integration/plugin/authz/authz_plugin_test.go b/components/engine/integration/plugin/authz/authz_plugin_test.go index 5bca6c138d..5585d93024 100644 --- a/components/engine/integration/plugin/authz/authz_plugin_test.go +++ b/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}) diff --git a/components/engine/pkg/authorization/authz.go b/components/engine/pkg/authorization/authz.go index fd87e9bef0..a1edbcd89d 100644 --- a/components/engine/pkg/authorization/authz.go +++ b/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 diff --git a/components/engine/pkg/authorization/authz_unix_test.go b/components/engine/pkg/authorization/authz_unix_test.go index d9c7c70e77..cfdb9a0039 100644 --- a/components/engine/pkg/authorization/authz_unix_test.go +++ b/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) diff --git a/components/engine/pkg/authorization/response.go b/components/engine/pkg/authorization/response.go index 647ff9ff6e..6b674bc295 100644 --- a/components/engine/pkg/authorization/response.go +++ b/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 diff --git a/components/engine/vendor.conf b/components/engine/vendor.conf index 9237560c75..11176a14a3 100644 --- a/components/engine/vendor.conf +++ b/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 diff --git a/components/engine/vendor/github.com/docker/libnetwork/drivers/overlay/ostweaks_linux.go b/components/engine/vendor/github.com/docker/libnetwork/drivers/overlay/ostweaks_linux.go index c00243ad2c..68f1ee9cee 100644 --- a/components/engine/vendor/github.com/docker/libnetwork/drivers/overlay/ostweaks_linux.go +++ b/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) } } } diff --git a/components/engine/vendor/github.com/docker/libnetwork/networkdb/delegate.go b/components/engine/vendor/github.com/docker/libnetwork/networkdb/delegate.go index c308fde795..1a57de3de6 100644 --- a/components/engine/vendor/github.com/docker/libnetwork/networkdb/delegate.go +++ b/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)