Align header reads

docker.go

@@ -667,17 +667,22 @@ func (c *DockerContainer) StartLogProducer(ctx context.Context) error {
 				return
 			default:
 				h := make([]byte, 8)
-				_, err := r.Read(h)
+				_, err := io.ReadFull(r, h)
 				if err != nil {
 					// proper type matching requires https://go-review.googlesource.com/c/go/+/250357/ (go 1.16)
 					if strings.Contains(err.Error(), "use of closed network connection") {
 						now := time.Now()
 						since = fmt.Sprintf("%d.%09d", now.Unix(), int64(now.Nanosecond()))
 						goto BEGIN
 					}
-					// this explicitly ignores errors
-					// because we want to keep procesing even if one of our reads fails
-					continue
+					if errors.Is(err, context.DeadlineExceeded) {
+						// Probably safe to continue here
+						continue
+					}
+					_, _ = fmt.Fprintf(os.Stderr, "container log error: %+v", err)
+					// if we would continue here, the next header-read will result into random data...
+					<-stop
+					return
 				}
 
 				count := binary.BigEndian.Uint32(h[4:])
@@ -695,11 +700,17 @@ func (c *DockerContainer) StartLogProducer(ctx context.Context) error {
 				logTypes := []string{"", StdoutLog, StderrLog}
 
 				b := make([]byte, count)
-				_, err = r.Read(b)
+				_, err = io.ReadFull(r, b)
 				if err != nil {
 					// TODO: add-logger: use logger to log out this error
 					_, _ = fmt.Fprintf(os.Stderr, "error occurred reading log with known length %s", err.Error())
-					continue
+					if errors.Is(err, context.DeadlineExceeded) {
+						// Probably safe to continue here
+						continue
+					}
+					// we can not continie here as the next read most likely will not be the next header
+					<-stop
+					return
 				}
 				for _, c := range c.consumers {
 					c.Accept(Log{