Repository: open-telemetry/opamp-go
Pull request #61 (2022-04-07 20:13:20)

Implement plain HTTP Transport in OpAMPServer
This adds plain HTTP request support to OpAMPServer implementation.
Implements spec change: https://github.com/open-telemetry/opamp-spec/pull/70

tigrannajaryan: @open-telemetry/opamp-go-approvers please review.

Patch:
diff --git a/internal/examples/server/opampsrv/opampsrv.go b/internal/examples/server/opampsrv/opampsrv.go
index cd04614..d37f742 100644
--- a/internal/examples/server/opampsrv/opampsrv.go
+++ b/internal/examples/server/opampsrv/opampsrv.go
@@ -53,7 +53,7 @@ func (srv *Server) onDisconnect(conn types.Connection) {
 	srv.agents.RemoveConnection(conn)
 }
 
-func (srv *Server) onMessage(conn types.Connection, msg *protobufs.AgentToServer) {
+func (srv *Server) onMessage(conn types.Connection, msg *protobufs.AgentToServer) *protobufs.ServerToAgent {
 	instanceId := data.InstanceId(msg.InstanceUid)
 
 	agent := srv.agents.FindOrCreateAgent(instanceId, conn)
@@ -69,5 +69,5 @@ func (srv *Server) onMessage(conn types.Connection, msg *protobufs.AgentToServer
 	}
 
 	// Send the response back to the agent.
-	agent.SendToAgent(response)
+	return response
 }
diff --git a/server/callbacks.go b/server/callbacks.go
index 7864701..46619e5 100644
--- a/server/callbacks.go
+++ b/server/callbacks.go
@@ -10,7 +10,7 @@ import (
 type CallbacksStruct struct {
 	OnConnectingFunc      func(request *http.Request) types.ConnectionResponse
 	OnConnectedFunc       func(conn types.Connection)
-	OnMessageFunc         func(conn types.Connection, message *protobufs.AgentToServer)
+	OnMessageFunc         func(conn types.Connection, message *protobufs.AgentToServer) *protobufs.ServerToAgent
 	OnConnectionCloseFunc func(conn types.Connection)
 }
 
@@ -29,9 +29,14 @@ func (c CallbacksStruct) OnConnected(conn types.Connection) {
 	}
 }
 
-func (c CallbacksStruct) OnMessage(conn types.Connection, message *protobufs.AgentToServer) {
+func (c CallbacksStruct) OnMessage(conn types.Connection, message *protobufs.AgentToServer) *protobufs.ServerToAgent {
 	if c.OnMessageFunc != nil {
-		c.OnMessageFunc(conn, message)
+		return c.OnMessageFunc(conn, message)
+	} else {
+		// We will send an empty response since there is no user-defined callback to handle it.
+		return &protobufs.ServerToAgent{
+			InstanceUid: message.InstanceUid,
+		}
 	}
 }
 
diff --git a/server/httpconnection.go b/server/httpconnection.go
new file mode 100644
index 0000000..1ed1b29
--- /dev/null
+++ b/server/httpconnection.go
@@ -0,0 +1,32 @@
+package server
+
+import (
+	"context"
+	"errors"
+	"net"
+
+	"github.com/open-telemetry/opamp-go/protobufs"
+	"github.com/open-telemetry/opamp-go/server/types"
+)
+
+var errInvalidHTTPConnection = errors.New("cannot Send() over HTTP connection")
+
+// httpConnection represents an OpAMP connection over a plain HTTP connection.
+// Only one response is possible to send when using plain HTTP connection
+// and that response will be sent by OpAMP server's HTTP request handler after the
+// onMessage callback returns.
+type httpConnection struct {
+	conn net.Conn
+}
+
+func (c httpConnection) RemoteAddr() net.Addr {
+	return c.conn.RemoteAddr()
+}
+
+var _ types.Connection = (*httpConnection)(nil)
+
+func (c httpConnection) Send(_ context.Context, _ *protobufs.ServerToAgent) error {
+	// Send() should not be called for plain HTTP connection. Instead, the response will
+	// be sent after the onMessage callback returns.
+	return errInvalidHTTPConnection
+}
diff --git a/server/httpconnectioncontext.go b/server/httpconnectioncontext.go
new file mode 100644
index 0000000..f35803a
--- /dev/null
+++ b/server/httpconnectioncontext.go
@@ -0,0 +1,23 @@
+package server
+
+import (
+	"context"
+	"net"
+	"net/http"
+)
+
+type connContextKeyType struct {
+	key string
+}
+
+var connContextKey = connContextKeyType{key: "httpconn"}
+
+func contextWithConn(ctx context.Context, c net.Conn) context.Context {
+	// Create a new context that stores the net.Conn. For use as ConnContext func
+	// of http.Server to remember the connection in the context.
+	return context.WithValue(ctx, connContextKey, c)
+}
+func connFromRequest(r *http.Request) net.Conn {
+	// Extract the net.Conn from the context of the specified http.Request.
+	return r.Context().Value(connContextKey).(net.Conn)
+}
diff --git a/server/serverimpl.go b/server/serverimpl.go
index c2c6eb6..5fb44e8 100644
--- a/server/serverimpl.go
+++ b/server/serverimpl.go
@@ -3,6 +3,7 @@ package server
 import (
 	"context"
 	"errors"
+	"io"
 	"net"
 	"net/http"
 
@@ -19,6 +20,8 @@ var (
 )
 
 const defaultOpAMPPath = "/v1/opamp"
+const headerContentType = "Content-Type"
+const contentTypeProtobuf = "application/x-protobuf"
 
 type server struct {
 	logger   types.Logger
@@ -69,9 +72,10 @@ func (s *server) Start(settings StartSettings) error {
 	mux.HandleFunc(path, s.httpHandler)
 
 	hs := &http.Server{
-		Handler:   mux,
-		Addr:      settings.ListenEndpoint,
-		TLSConfig: settings.TLSConfig,
+		Handler:     mux,
+		Addr:        settings.ListenEndpoint,
+		TLSConfig:   settings.TLSConfig,
+		ConnContext: contextWithConn,
 	}
 	s.httpServer = hs
 
@@ -143,7 +147,15 @@ func (s *server) httpHandler(w http.ResponseWriter, req *http.Request) {
 		}
 	}
 
-	// HTTP connection is accepted. Upgrade it to WebSocket.
+	// HTTP connection is accepted. Check if it is a plain HTTP request.
+
+	if req.Header.Get(headerContentType) == contentTypeProtobuf {
+		// Yes, a plain HTTP request.
+		s.handlePlainHTTPRequest(req, w)
+		return
+	}
+
+	// No, it is a WebSocket. Upgrade it.
 	conn, err := s.wsUpgrader.Upgrade(w, req, nil)
 	if err != nil {
 		s.logger.Errorf("Cannot upgrade HTTP connection to WebSocket: %v", err)
@@ -156,11 +168,16 @@ func (s *server) httpHandler(w http.ResponseWriter, req *http.Request) {
 }
 
 func (s *server) handleWSConnection(wsConn *websocket.Conn) {
-	agentConn := connection{wsConn: wsConn}
+	agentConn := wsConnection{wsConn: wsConn}
 
 	defer func() {
 		// Close the connection when all is done.
-		defer wsConn.Close()
+		defer func() {
+			err := wsConn.Close()
+			if err != nil {
+				s.logger.Errorf("error closing the WebSocket connection: %v", err)
+			}
+		}()
 
 		if s.settings.Callbacks != nil {
 			s.settings.Callbacks.OnConnectionClose(agentConn)
@@ -198,7 +215,73 @@ func (s *server) handleWSConnection(wsConn *websocket.Conn) {
 		}
 
 		if s.settings.Callbacks != nil {
-			s.settings.Callbacks.OnMessage(agentConn, &request)
+			response := s.settings.Callbacks.OnMessage(agentConn, &request)
+			if response.InstanceUid == "" {
+				response.InstanceUid = request.InstanceUid
+			}
+			err = agentConn.Send(context.Background(), response)
+			if err != nil {
+				s.logger.Errorf("Cannot send message to WebSocket: %v", err)
+			}
 		}
 	}
 }
+
+func (s *server) handlePlainHTTPRequest(req *http.Request, w http.ResponseWriter) {
+	bytes, err := io.ReadAll(req.Body)
+	if err != nil {
+		s.logger.Debugf("Cannot read HTTP body: %v", err)
+		w.WriteHeader(http.StatusBadRequest)
+		return
+	}
+
+	// Decode the message as a Protobuf message.
+	var request protobufs.AgentToServer
+	err = proto.Unmarshal(bytes, &request)
+	if err != nil {
+		s.logger.Debugf("Cannot decode message from HTTP Body: %v", err)
+		w.WriteHeader(http.StatusBadRequest)
+		return
+	}
+
+	agentConn := httpConnection{
+		conn: connFromRequest(req),
+	}
+
+	if s.settings.Callbacks == nil {
+		w.WriteHeader(http.StatusInternalServerError)
+		return
+	}
+
+	s.settings.Callbacks.OnConnected(agentConn)
+
+	defer func() {
+		// Indicate via the callback that the OpAMP Connection is closed. From OpAMP
+		// perspective the connection represented by this http request
+		// is closed. It is not possible to send or receive more OpAMP messages
+		// via this agentConn.
+		s.settings.Callbacks.OnConnectionClose(agentConn)
+	}()
+
+	response := s.settings.Callbacks.OnMessage(agentConn, &request)
+
+	// Set the InstanceUid if it is not set by the callback.
+	if response.InstanceUid == "" {
+		response.InstanceUid = request.InstanceUid
+	}
+
+	// Marshal the response.
+	bytes, err = proto.Marshal(response)
+	if err != nil {
+		w.WriteHeader(http.StatusInternalServerError)
+		return
+	}
+
+	// Send the response.
+	w.Header().Set(headerContentType, contentTypeProtobuf)
+	_, err = w.Write(bytes)
+
+	if err != nil {
+		s.logger.Debugf("Cannot send HTTP response: %v", err)
+	}
+}
diff --git a/server/types/callbacks.go b/server/types/callbacks.go
index 0305d4a..ff4fda0 100644
--- a/server/types/callbacks.go
+++ b/server/types/callbacks.go
@@ -29,16 +29,17 @@ type Callbacks interface {
 	//   HTTPResponseHeader may be optionally set (e.g. "Retry-After: 30").
 	OnConnecting(request *http.Request) ConnectionResponse
 
-	// OnConnected is called when the WebSocket connection is successfully established
-	// after OnConnecting() returns and the HTTP connection is upgraded to WebSocket.
+	// OnConnected is called when and incoming OpAMP connection is successfully
+	// established after OnConnecting() returns.
 	OnConnected(conn Connection)
 
 	// OnMessage is called when a message is received from the connection. Can happen
-	// only after OnConnected().
-	OnMessage(conn Connection, message *protobufs.AgentToServer)
+	// only after OnConnected(). Must return a ServerToAgent message that will be sent
+	// as a response to the agent.
+	// For plain HTTP requests once OnMessage returns and the response is sent
+	// to the agent the OnConnectionClose message will be called immediately.
+	OnMessage(conn Connection, message *protobufs.AgentToServer) *protobufs.ServerToAgent
 
-	// OnConnectionClose is called when the WebSocket connection is closed.
-	// Typically, preceded by OnDisconnect() unless the client misbehaves or the
-	// connection is lost.
+	// OnConnectionClose is called when the OpAMP connection is closed.
 	OnConnectionClose(conn Connection)
 }
diff --git a/server/types/connection.go b/server/types/connection.go
index 26da474..22e5ac8 100644
--- a/server/types/connection.go
+++ b/server/types/connection.go
@@ -7,13 +7,15 @@ import (
 	"github.com/open-telemetry/opamp-go/protobufs"
 )
 
-// Connection represents one OpAMP WebSocket connections.
+// Connection represents one OpAMP connection.
 // The implementation MUST be a comparable type so that it can be used as a map key.
 type Connection interface {
 	// RemoteAddr returns the remote network address of the connection.
 	RemoteAddr() net.Addr
 
 	// Send a message. Should not be called concurrently for the same Connection instance.
+	// Can be called only for WebSocket connections. Will return an error for plain HTTP
+	// connections.
 	// Blocks until the message is sent.
 	// Should return as soon as possible if the ctx is cancelled.
 	Send(ctx context.Context, message *protobufs.ServerToAgent) error
diff --git a/server/connection.go b/server/wsconnection.go
similarity index 62%
rename from server/connection.go
rename to server/wsconnection.go
index 9f62c9e..6d90f75 100644
--- a/server/connection.go
+++ b/server/wsconnection.go
@@ -11,17 +11,18 @@ import (
 	"github.com/open-telemetry/opamp-go/server/types"
 )
 
-type connection struct {
+// wsConnection represents a persistent OpAMP connection over a WebSocket.
+type wsConnection struct {
 	wsConn *websocket.Conn
 }
 
-var _ types.Connection = (*connection)(nil)
+var _ types.Connection = (*wsConnection)(nil)
 
-func (c connection) RemoteAddr() net.Addr {
+func (c wsConnection) RemoteAddr() net.Addr {
 	return c.wsConn.RemoteAddr()
 }
 
-func (c connection) Send(ctx context.Context, message *protobufs.ServerToAgent) error {
+func (c wsConnection) Send(_ context.Context, message *protobufs.ServerToAgent) error {
 	bytes, err := proto.Marshal(message)
 	if err != nil {
 		return err


Test patch:
diff --git a/server/serverimpl_test.go b/server/serverimpl_test.go
index b0da248..4fa9aff 100644
--- a/server/serverimpl_test.go
+++ b/server/serverimpl_test.go
@@ -1,7 +1,9 @@
 package server
 
 import (
+	"bytes"
 	"context"
+	"io"
 	"net/http"
 	"net/http/httptest"
 	"sync/atomic"
@@ -130,7 +132,7 @@ func TestServerReceiveSendMessage(t *testing.T) {
 		OnConnectingFunc: func(request *http.Request) types.ConnectionResponse {
 			return types.ConnectionResponse{Accept: true}
 		},
-		OnMessageFunc: func(conn types.Connection, message *protobufs.AgentToServer) {
+		OnMessageFunc: func(conn types.Connection, message *protobufs.AgentToServer) *protobufs.ServerToAgent {
 			// Remember received message.
 			rcvMsg.Store(message)
 
@@ -139,8 +141,7 @@ func TestServerReceiveSendMessage(t *testing.T) {
 				InstanceUid:  message.InstanceUid,
 				Capabilities: protobufs.ServerCapabilities_AcceptsStatus,
 			}
-			err := conn.Send(context.Background(), &response)
-			assert.NoError(t, err)
+			return &response
 		},
 	}
 
@@ -182,6 +183,72 @@ func TestServerReceiveSendMessage(t *testing.T) {
 	assert.EqualValues(t, protobufs.ServerCapabilities_AcceptsStatus, response.Capabilities)
 }
 
+func TestServerReceiveSendMessagePlainHTTP(t *testing.T) {
+	var rcvMsg atomic.Value
+	var onConnectedCalled, onCloseCalled int32
+	callbacks := CallbacksStruct{
+		OnConnectingFunc: func(request *http.Request) types.ConnectionResponse {
+			return types.ConnectionResponse{Accept: true}
+		},
+		OnConnectedFunc: func(conn types.Connection) {
+			atomic.StoreInt32(&onConnectedCalled, 1)
+		},
+		OnMessageFunc: func(conn types.Connection, message *protobufs.AgentToServer) *protobufs.ServerToAgent {
+			// Remember received message.
+			rcvMsg.Store(message)
+
+			// Send a response.
+			response := protobufs.ServerToAgent{
+				InstanceUid:  message.InstanceUid,
+				Capabilities: protobufs.ServerCapabilities_AcceptsStatus,
+			}
+			return &response
+		},
+		OnConnectionCloseFunc: func(conn types.Connection) {
+			atomic.StoreInt32(&onCloseCalled, 1)
+		},
+	}
+
+	// Start a server.
+	settings := &StartSettings{Settings: Settings{Callbacks: callbacks}}
+	srv := startServer(t, settings)
+	defer srv.Stop(context.Background())
+
+	// Send a message to the server.
+	sendMsg := protobufs.AgentToServer{
+		InstanceUid: "12345678",
+	}
+	b, err := proto.Marshal(&sendMsg)
+	require.NoError(t, err)
+	resp, err := http.Post("http://"+settings.ListenEndpoint+settings.ListenPath, contentTypeProtobuf, bytes.NewReader(b))
+	require.NoError(t, err)
+
+	// Wait until server receives the message.
+	eventually(t, func() bool { return rcvMsg.Load() != nil })
+	assert.True(t, atomic.LoadInt32(&onConnectedCalled) == 1)
+
+	// Verify the received message is what was sent.
+	assert.True(t, proto.Equal(rcvMsg.Load().(proto.Message), &sendMsg))
+
+	// Read server's response.
+	b, err = io.ReadAll(resp.Body)
+	require.NoError(t, err)
+
+	assert.EqualValues(t, http.StatusOK, resp.StatusCode)
+	assert.EqualValues(t, contentTypeProtobuf, resp.Header.Get(headerContentType))
+
+	// Decode the response.
+	var response protobufs.ServerToAgent
+	err = proto.Unmarshal(b, &response)
+	require.NoError(t, err)
+
+	// Verify the response.
+	assert.EqualValues(t, sendMsg.InstanceUid, response.InstanceUid)
+	assert.EqualValues(t, protobufs.ServerCapabilities_AcceptsStatus, response.Capabilities)
+
+	eventually(t, func() bool { return atomic.LoadInt32(&onCloseCalled) == 1 })
+}
+
 func TestServerAttachAcceptConnection(t *testing.T) {
 	connectedCalled := int32(0)
 	connectionCloseCalled := int32(0)