Refactor common bridge protocol code for reuse

- Move common bridge protocol definitions to subpackage
under internal/gcs
- Move helper functions to internal/bridgeutils pkg
so that they can be used by gcs-sidecar as well

Signed-off-by: Kirtana Ashok <[email protected]>

Author: kiashok

internal/bridgeutils/commonutils/utilities.go

@@ -0,0 +1,83 @@
+package commonutils
+
+import (
+	"encoding/json"
+	"fmt"
+	"io"
+	"math"
+	"strconv"
+
+	"github.com/Microsoft/hcsshim/internal/bridgeutils/gcserr"
+	"github.com/sirupsen/logrus"
+)
+
+type ErrorRecord struct {
+	Result       int32 // HResult
+	Message      string
+	StackTrace   string `json:",omitempty"`
+	ModuleName   string
+	FileName     string
+	Line         uint32
+	FunctionName string `json:",omitempty"`
+}
+
+// UnmarshalJSONWithHresult unmarshals the given data into the given interface, and
+// wraps any error returned in an HRESULT error.
+func UnmarshalJSONWithHresult(data []byte, v interface{}) error {
+	if err := json.Unmarshal(data, v); err != nil {
+		return gcserr.WrapHresult(err, gcserr.HrVmcomputeInvalidJSON)
+	}
+	return nil
+}
+
+// DecodeJSONWithHresult decodes the JSON from the given reader into the given
+// interface, and wraps any error returned in an HRESULT error.
+func DecodeJSONWithHresult(r io.Reader, v interface{}) error {
+	if err := json.NewDecoder(r).Decode(v); err != nil {
+		return gcserr.WrapHresult(err, gcserr.HrVmcomputeInvalidJSON)
+	}
+	return nil
+}
+
+func SetErrorForResponseBaseUtil(errForResponse error, moduleName string) (hresult gcserr.Hresult, errorMessage string, newRecord ErrorRecord) {
+	errorMessage = errForResponse.Error()
+	stackString := ""
+	fileName := ""
+	// We use -1 as a sentinel if no line number found (or it cannot be parsed),
+	// but that will ultimately end up as [math.MaxUint32], so set it to that explicitly.
+	// (Still keep using -1 for backwards compatibility ...)
+	lineNumber := uint32(math.MaxUint32)
+	functionName := ""
+	if stack := gcserr.BaseStackTrace(errForResponse); stack != nil {
+		bottomFrame := stack[0]
+		stackString = fmt.Sprintf("%+v", stack)
+		fileName = fmt.Sprintf("%s", bottomFrame)
+		lineNumberStr := fmt.Sprintf("%d", bottomFrame)
+		if n, err := strconv.ParseUint(lineNumberStr, 10, 32); err == nil {
+			lineNumber = uint32(n)
+		} else {
+			logrus.WithFields(logrus.Fields{
+				"line-number":   lineNumberStr,
+				logrus.ErrorKey: err,
+			}).Error("opengcs::bridge::setErrorForResponseBase - failed to parse line number, using -1 instead")
+		}
+		functionName = fmt.Sprintf("%n", bottomFrame)
+	}
+	hresult, err := gcserr.GetHresult(errForResponse)
+	if err != nil {
+		// Default to using the generic failure HRESULT.
+		hresult = gcserr.HrFail
+	}
+
+	newRecord = ErrorRecord{
+		Result:       int32(hresult),
+		Message:      errorMessage,
+		StackTrace:   stackString,
+		ModuleName:   moduleName,
+		FileName:     fileName,
+		Line:         lineNumber,
+		FunctionName: functionName,
+	}
+
+	return hresult, errorMessage, newRecord
+}

internal/guest/gcserr/errors.go renamed to internal/bridgeutils/gcserr/errors.go

@@ -19,6 +19,7 @@ import (
 	"go.opencensus.io/trace"
 	"golang.org/x/sys/windows"
 
+	"github.com/Microsoft/hcsshim/internal/gcs/prot"
 	"github.com/Microsoft/hcsshim/internal/log"
 	"github.com/Microsoft/hcsshim/internal/oc"
 )
@@ -38,16 +39,16 @@ const (
 )
 
 type requestMessage interface {
-	Base() *requestBase
+	Base() *prot.RequestBase
 }
 
 type responseMessage interface {
-	Base() *responseBase
+	Base() *prot.ResponseBase
 }
 
 // rpc represents an outstanding rpc request to the guest
 type rpc struct {
-	proc    rpcProc
+	proc    prot.RpcProc
 	id      int64
 	req     requestMessage
 	resp    responseMessage
@@ -80,7 +81,7 @@ const (
 	bridgeFailureTimeout = time.Minute * 5
 )
 
-type notifyFunc func(*containerNotification) error
+type notifyFunc func(*prot.ContainerNotification) error
 
 // newBridge returns a bridge on `conn`. It calls `notify` when a
 // notification message arrives from the guest. It logs transport errors and
@@ -143,7 +144,7 @@ func (brdg *bridge) Wait() error {
 // AsyncRPC sends an RPC request to the guest but does not wait for a response.
 // If the message cannot be sent before the context is done, then an error is
 // returned.
-func (brdg *bridge) AsyncRPC(ctx context.Context, proc rpcProc, req requestMessage, resp responseMessage) (*rpc, error) {
+func (brdg *bridge) AsyncRPC(ctx context.Context, proc prot.RpcProc, req requestMessage, resp responseMessage) (*rpc, error) {
 	call := &rpc{
 		ch:   make(chan struct{}),
 		proc: proc,
@@ -224,7 +225,7 @@ func (call *rpc) Wait() {
 // If allowCancel is set and the context becomes done, returns an error without
 // waiting for a response. Avoid this on messages that are not idempotent or
 // otherwise safe to ignore the response of.
-func (brdg *bridge) RPC(ctx context.Context, proc rpcProc, req requestMessage, resp responseMessage, allowCancel bool) error {
+func (brdg *bridge) RPC(ctx context.Context, proc prot.RpcProc, req requestMessage, resp responseMessage, allowCancel bool) error {
 	call, err := brdg.AsyncRPC(ctx, proc, req, resp)
 	if err != nil {
 		return err
@@ -261,7 +262,7 @@ func (brdg *bridge) recvLoopRoutine() {
 	}
 }
 
-func readMessage(r io.Reader) (int64, msgType, []byte, error) {
+func readMessage(r io.Reader) (int64, prot.MsgType, []byte, error) {
 	_, span := oc.StartSpan(context.Background(), "bridge receive read message", oc.WithClientSpanKind)
 	defer span.End()
 
@@ -270,7 +271,7 @@ func readMessage(r io.Reader) (int64, msgType, []byte, error) {
 	if err != nil {
 		return 0, 0, nil, fmt.Errorf("header read: %w", err)
 	}
-	typ := msgType(binary.LittleEndian.Uint32(h[hdrOffType:]))
+	typ := prot.MsgType(binary.LittleEndian.Uint32(h[hdrOffType:]))
 	n := binary.LittleEndian.Uint32(h[hdrOffSize:])
 	id := int64(binary.LittleEndian.Uint64(h[hdrOffID:]))
 	span.AddAttributes(
@@ -311,8 +312,8 @@ func (brdg *bridge) recvLoop() error {
 			"type":       typ.String(),
 			"message-id": id}).Trace("bridge receive")
 
-		switch typ & msgTypeMask {
-		case msgTypeResponse:
+		switch typ & prot.MsgTypeMask {
+		case prot.MsgTypeResponse:
 			// Find the request associated with this response.
 			brdg.mu.Lock()
 			call := brdg.rpcs[id]
@@ -344,11 +345,11 @@ func (brdg *bridge) recvLoop() error {
 				return err
 			}
 
-		case msgTypeNotify:
-			if typ != notifyContainer|msgTypeNotify {
+		case prot.MsgTypeNotify:
+			if typ != prot.NotifyContainer|prot.MsgTypeNotify {
 				return fmt.Errorf("bridge received unknown unknown notification message %s", typ)
 			}
-			var ntf containerNotification
+			var ntf prot.ContainerNotification
 			ntf.ResultInfo.Value = &json.RawMessage{}
 			err := json.Unmarshal(b, &ntf)
 			if err != nil {
@@ -383,7 +384,7 @@ func (brdg *bridge) sendLoop() {
 	}
 }
 
-func (brdg *bridge) writeMessage(buf *bytes.Buffer, enc *json.Encoder, typ msgType, id int64, req interface{}) error {
+func (brdg *bridge) writeMessage(buf *bytes.Buffer, enc *json.Encoder, typ prot.MsgType, id int64, req interface{}) error {
 	var err error
 	_, span := oc.StartSpan(context.Background(), "bridge send", oc.WithClientSpanKind)
 	defer span.End()
@@ -408,9 +409,9 @@ func (brdg *bridge) writeMessage(buf *bytes.Buffer, enc *json.Encoder, typ msgTy
 		b := buf.Bytes()[hdrSize:]
 		switch typ {
 		// container environment vars are in rpCreate for linux; rpcExecuteProcess for windows
-		case msgType(rpcCreate) | msgTypeRequest:
+		case prot.MsgType(prot.RpcCreate) | prot.MsgTypeRequest:
 			b, err = log.ScrubBridgeCreate(b)
-		case msgType(rpcExecuteProcess) | msgTypeRequest:
+		case prot.MsgType(prot.RpcExecuteProcess) | prot.MsgTypeRequest:
 			b, err = log.ScrubBridgeExecProcess(b)
 		}
 		if err != nil {
@@ -443,7 +444,7 @@ func (brdg *bridge) sendRPC(buf *bytes.Buffer, enc *json.Encoder, call *rpc) err
 	brdg.rpcs[id] = call
 	brdg.nextID++
 	brdg.mu.Unlock()
-	typ := msgType(call.proc) | msgTypeRequest
+	typ := prot.MsgType(call.proc) | prot.MsgTypeRequest
 	err := brdg.writeMessage(buf, enc, typ, id, call.req)
 	if err != nil {
 		// Try to reclaim this request and fail it.

internal/gcs/bridge.go

@@ -13,6 +13,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/Microsoft/hcsshim/internal/gcs/prot"
 	"github.com/sirupsen/logrus"
 )
 
@@ -33,7 +34,7 @@ func pipeConn() (*stitched, *stitched) {
 	return &stitched{r1, w2}, &stitched{r2, w1}
 }
 
-func sendMessage(t *testing.T, w io.Writer, typ msgType, id int64, msg []byte) {
+func sendMessage(t *testing.T, w io.Writer, typ prot.MsgType, id int64, msg []byte) {
 	t.Helper()
 	var h [16]byte
 	binary.LittleEndian.PutUint32(h[:], uint32(typ))
@@ -63,18 +64,18 @@ func reflector(t *testing.T, rw io.ReadWriteCloser, delay time.Duration) {
 			return
 		}
 		time.Sleep(delay) // delay is used to test timeouts (when non-zero)
-		typ ^= msgTypeResponse ^ msgTypeRequest
+		typ ^= prot.MsgTypeResponse ^ prot.MsgTypeRequest
 		sendMessage(t, rw, typ, id, msg)
 	}
 }
 
 type testReq struct {
-	requestBase
+	prot.RequestBase
 	X, Y int
 }
 
 type testResp struct {
-	responseBase
+	prot.ResponseBase
 	X, Y int
 }
 
@@ -92,7 +93,7 @@ func TestBridgeRPC(t *testing.T) {
 	defer b.Close()
 	req := testReq{X: 5}
 	var resp testResp
-	err := b.RPC(context.Background(), rpcCreate, &req, &resp, false)
+	err := b.RPC(context.Background(), prot.RpcCreate, &req, &resp, false)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -107,7 +108,7 @@ func TestBridgeRPCResponseTimeout(t *testing.T) {
 	b.Timeout = time.Millisecond * 100
 	req := testReq{X: 5}
 	var resp testResp
-	err := b.RPC(context.Background(), rpcCreate, &req, &resp, false)
+	err := b.RPC(context.Background(), prot.RpcCreate, &req, &resp, false)
 	if err == nil || !strings.Contains(err.Error(), "bridge closed") {
 		t.Fatalf("expected bridge disconnection, got %s", err)
 	}
@@ -121,7 +122,7 @@ func TestBridgeRPCContextDone(t *testing.T) {
 	defer cancel()
 	req := testReq{X: 5}
 	var resp testResp
-	err := b.RPC(ctx, rpcCreate, &req, &resp, true)
+	err := b.RPC(ctx, prot.RpcCreate, &req, &resp, true)
 	if err != context.DeadlineExceeded { //nolint:errorlint
 		t.Fatalf("expected deadline exceeded, got %s", err)
 	}
@@ -135,7 +136,7 @@ func TestBridgeRPCContextDoneNoCancel(t *testing.T) {
 	defer cancel()
 	req := testReq{X: 5}
 	var resp testResp
-	err := b.RPC(ctx, rpcCreate, &req, &resp, false)
+	err := b.RPC(ctx, prot.RpcCreate, &req, &resp, false)
 	if err == nil || !strings.Contains(err.Error(), "bridge closed") {
 		t.Fatalf("expected bridge disconnection, got %s", err)
 	}
@@ -145,13 +146,13 @@ func TestBridgeRPCBridgeClosed(t *testing.T) {
 	b := startReflectedBridge(t, 0)
 	eerr := errors.New("forcibly terminated")
 	b.kill(eerr)
-	err := b.RPC(context.Background(), rpcCreate, nil, nil, false)
+	err := b.RPC(context.Background(), prot.RpcCreate, nil, nil, false)
 	if err != eerr { //nolint:errorlint
 		t.Fatal("unexpected: ", err)
 	}
 }
 
-func sendJSON(t *testing.T, w io.Writer, typ msgType, id int64, msg interface{}) error {
+func sendJSON(t *testing.T, w io.Writer, typ prot.MsgType, id int64, msg interface{}) error {
 	t.Helper()
 	msgb, err := json.Marshal(msg)
 	if err != nil {
@@ -161,7 +162,7 @@ func sendJSON(t *testing.T, w io.Writer, typ msgType, id int64, msg interface{})
 	return nil
 }
 
-func notifyThroughBridge(t *testing.T, typ msgType, msg interface{}, fn notifyFunc) error {
+func notifyThroughBridge(t *testing.T, typ prot.MsgType, msg interface{}, fn notifyFunc) error {
 	t.Helper()
 	s, c := pipeConn()
 	b := newBridge(s, fn, logrus.NewEntry(logrus.StandardLogger()))
@@ -176,9 +177,9 @@ func notifyThroughBridge(t *testing.T, typ msgType, msg interface{}, fn notifyFu
 }
 
 func TestBridgeNotify(t *testing.T) {
-	ntf := &containerNotification{Operation: "testing"}
+	ntf := &prot.ContainerNotification{Operation: "testing"}
 	recvd := false
-	err := notifyThroughBridge(t, msgTypeNotify|notifyContainer, ntf, func(nntf *containerNotification) error {
+	err := notifyThroughBridge(t, prot.MsgTypeNotify|prot.NotifyContainer, ntf, func(nntf *prot.ContainerNotification) error {
 		if !reflect.DeepEqual(ntf, nntf) {
 			t.Errorf("%+v != %+v", ntf, nntf)
 		}
@@ -194,9 +195,9 @@ func TestBridgeNotify(t *testing.T) {
 }
 
 func TestBridgeNotifyFailure(t *testing.T) {
-	ntf := &containerNotification{Operation: "testing"}
+	ntf := &prot.ContainerNotification{Operation: "testing"}
 	errMsg := "notify should have failed"
-	err := notifyThroughBridge(t, msgTypeNotify|notifyContainer, ntf, func(nntf *containerNotification) error {
+	err := notifyThroughBridge(t, prot.MsgTypeNotify|prot.NotifyContainer, ntf, func(nntf *prot.ContainerNotification) error {
 		return errors.New(errMsg)
 	})
 	if err == nil || !strings.Contains(err.Error(), errMsg) {