From f51e0d0597757da72fd7a10085998052684e396b Mon Sep 17 00:00:00 2001
From: Jhon Honce <jhonce@redhat.com>
Date: Wed, 20 May 2020 14:24:18 -0700
Subject: V2 enable remote logs and testing

* wire up bindings and handler for obtaining logs remotely
* enable debug logging from podman in e2e test using DEBUG and
  DEBUG_SERVICE env variables
* Fix error in streaming log frames
* enable remote logs test

Signed-off-by: Jhon Honce <jhonce@redhat.com>
---
 pkg/bindings/containers/containers.go |  7 ++--
 pkg/bindings/containers/logs.go       | 77 ++++++++++-------------------------
 pkg/bindings/test/containers_test.go  |  4 +-
 3 files changed, 27 insertions(+), 61 deletions(-)

(limited to 'pkg/bindings')

diff --git a/pkg/bindings/containers/containers.go b/pkg/bindings/containers/containers.go
index 1ed4919e0..39a077f36 100644
--- a/pkg/bindings/containers/containers.go
+++ b/pkg/bindings/containers/containers.go
@@ -23,7 +23,7 @@ import (
 )
 
 var (
-	ErrLostSync = errors.New("lost synchronization with attach multiplexed result")
+	ErrLostSync = errors.New("lost synchronization with multiplexed stream")
 )
 
 // List obtains a list of containers in local storage.  All parameters to this method are optional.
@@ -485,7 +485,7 @@ func Attach(ctx context.Context, nameOrId string, detachKeys *string, logs, stre
 					return err
 				}
 			case fd == 3:
-				return fmt.Errorf("error from daemon in stream: %s", frame)
+				return errors.New("error from service in stream: " + string(frame))
 			default:
 				return fmt.Errorf("unrecognized input header: %d", fd)
 			}
@@ -507,7 +507,7 @@ func DemuxHeader(r io.Reader, buffer []byte) (fd, sz int, err error) {
 
 	fd = int(buffer[0])
 	if fd < 0 || fd > 3 {
-		err = ErrLostSync
+		err = errors.Wrapf(ErrLostSync, fmt.Sprintf(`channel "%d" found, 0-3 supported`, fd))
 		return
 	}
 
@@ -528,7 +528,6 @@ func DemuxFrame(r io.Reader, buffer []byte, length int) (frame []byte, err error
 		err = io.ErrUnexpectedEOF
 		return
 	}
-
 	return buffer[0:length], nil
 }
 
diff --git a/pkg/bindings/containers/logs.go b/pkg/bindings/containers/logs.go
index b7ecb3c7e..20c8b4292 100644
--- a/pkg/bindings/containers/logs.go
+++ b/pkg/bindings/containers/logs.go
@@ -1,8 +1,9 @@
 package containers
 
 import (
+	"bytes"
 	"context"
-	"encoding/binary"
+	"fmt"
 	"io"
 	"net/http"
 	"net/url"
@@ -49,68 +50,34 @@ func Logs(ctx context.Context, nameOrID string, opts LogOptions, stdoutChan, std
 	if err != nil {
 		return err
 	}
+	defer response.Body.Close()
 
-	// read 8 bytes
-	// first byte determines stderr=2|stdout=1
-	// bytes 4-7 len(msg) in uint32
+	buffer := make([]byte, 1024)
 	for {
-		stream, msgSize, err := readHeader(response.Body)
+		fd, l, err := DemuxHeader(response.Body, buffer)
 		if err != nil {
-			// In case the server side closes up shop because !follow
-			if err == io.EOF {
-				break
+			if errors.Is(err, io.EOF) {
+				return nil
 			}
-			return errors.Wrap(err, "unable to read log header")
+			return err
 		}
-		msg, err := readMsg(response.Body, msgSize)
+		frame, err := DemuxFrame(response.Body, buffer, l)
 		if err != nil {
-			return errors.Wrap(err, "unable to read log message")
+			return err
 		}
-		if stream == 1 {
-			stdoutChan <- msg
-		} else {
-			stderrChan <- msg
-		}
-	}
-	return nil
-}
+		frame = bytes.Replace(frame[0:l], []byte{13}, []byte{10}, -1)
 
-func readMsg(r io.Reader, msgSize int) (string, error) {
-	var msg []byte
-	size := msgSize
-	for {
-		b := make([]byte, size)
-		_, err := r.Read(b)
-		if err != nil {
-			return "", err
-		}
-		msg = append(msg, b...)
-		if len(msg) == msgSize {
-			break
-		}
-		size = msgSize - len(msg)
-	}
-	return string(msg), nil
-}
-
-func readHeader(r io.Reader) (byte, int, error) {
-	var (
-		header []byte
-		size   = 8
-	)
-	for {
-		b := make([]byte, size)
-		_, err := r.Read(b)
-		if err != nil {
-			return 0, 0, err
-		}
-		header = append(header, b...)
-		if len(header) == 8 {
-			break
+		switch fd {
+		case 0:
+			stdoutChan <- string(frame)
+		case 1:
+			stdoutChan <- string(frame)
+		case 2:
+			stderrChan <- string(frame)
+		case 3:
+			return errors.New("error from service in stream: " + string(frame))
+		default:
+			return fmt.Errorf("unrecognized input header: %d", fd)
 		}
-		size = 8 - len(header)
 	}
-	stream := header[0]
-	msgSize := int(binary.BigEndian.Uint32(header[4:]) - 8)
-	return stream, msgSize, nil
 }
diff --git a/pkg/bindings/test/containers_test.go b/pkg/bindings/test/containers_test.go
index f725d1cf2..3b94b10eb 100644
--- a/pkg/bindings/test/containers_test.go
+++ b/pkg/bindings/test/containers_test.go
@@ -378,9 +378,9 @@ var _ = Describe("Podman containers ", func() {
 			containers.Logs(bt.conn, r.ID, opts, stdoutChan, nil)
 		}()
 		o := <-stdoutChan
-		o = strings.ReplaceAll(o, "\r", "")
+		o = strings.TrimSpace(o)
 		_, err = time.Parse(time.RFC1123Z, o)
-		Expect(err).To(BeNil())
+		Expect(err).ShouldNot(HaveOccurred())
 	})
 
 	It("podman top", func() {
-- 
cgit v1.2.3-54-g00ecf