Repository: open-telemetry/opamp-go
Pull request #33 (2021-12-06 21:57:06)

Begin basic server implementation
Contributes to https://github.com/open-telemetry/opamp-go/issues/32

This implements the OpAMPServer interface and adds some basic tests
to verify the functionality.

Also added ListenPath to server StartSettings since it is necessary
to implement the server.

TODO:
- Handle closing and waiting for existing connections.
- Wait for listening in Start().
- Decide what to do with OnDisconnect callback.


Patch:
diff --git a/client/clientimpl.go b/client/clientimpl.go
index 1f60458..64f41fc 100644
--- a/client/clientimpl.go
+++ b/client/clientimpl.go
@@ -15,6 +15,7 @@ import (
 
 	"github.com/open-telemetry/opamp-go/client/internal"
 	"github.com/open-telemetry/opamp-go/client/types"
+	sharedinternal "github.com/open-telemetry/opamp-go/internal"
 	"github.com/open-telemetry/opamp-go/internal/protobufs"
 )
 
@@ -62,7 +63,7 @@ const retryAfterHTTPHeader = "Retry-After"
 
 func New(logger types.Logger) *client {
 	if logger == nil {
-		logger = &nopLogger{}
+		logger = &sharedinternal.NopLogger{}
 	}
 
 	w := &client{
diff --git a/client/noplogger.go b/client/noplogger.go
deleted file mode 100644
index 61eea18..0000000
--- a/client/noplogger.go
+++ /dev/null
@@ -1,6 +0,0 @@
-package client
-
-type nopLogger struct{}
-
-func (l *nopLogger) Debugf(format string, v ...interface{}) {}
-func (l *nopLogger) Errorf(format string, v ...interface{}) {}
diff --git a/internal/noplogger.go b/internal/noplogger.go
new file mode 100644
index 0000000..a575c5d
--- /dev/null
+++ b/internal/noplogger.go
@@ -0,0 +1,6 @@
+package internal
+
+type NopLogger struct{}
+
+func (l *NopLogger) Debugf(format string, v ...interface{}) {}
+func (l *NopLogger) Errorf(format string, v ...interface{}) {}
diff --git a/makefile b/makefile
index aa4f4c4..54e7688 100644
--- a/makefile
+++ b/makefile
@@ -12,7 +12,7 @@ all: test
 
 .PHONY: test
 test:
-	go test ./...
+	go test -race ./...
 
 # Generate Protobuf Go files.
 .PHONY: gen-proto
diff --git a/server/connection.go b/server/connection.go
new file mode 100644
index 0000000..2b19de1
--- /dev/null
+++ b/server/connection.go
@@ -0,0 +1,25 @@
+package server
+
+import (
+	"context"
+
+	"github.com/gorilla/websocket"
+	"google.golang.org/protobuf/proto"
+
+	"github.com/open-telemetry/opamp-go/internal/protobufs"
+	"github.com/open-telemetry/opamp-go/server/types"
+)
+
+type connection struct {
+	wsConn *websocket.Conn
+}
+
+var _ types.Connection = (*connection)(nil)
+
+func (c connection) Send(ctx context.Context, message *protobufs.ServerToAgent) error {
+	bytes, err := proto.Marshal(message)
+	if err != nil {
+		return err
+	}
+	return c.wsConn.WriteMessage(websocket.BinaryMessage, bytes)
+}
diff --git a/server/internal/callbacks.go b/server/internal/callbacks.go
new file mode 100644
index 0000000..bd0e695
--- /dev/null
+++ b/server/internal/callbacks.go
@@ -0,0 +1,49 @@
+package internal
+
+import (
+	"net/http"
+
+	"github.com/open-telemetry/opamp-go/internal/protobufs"
+	"github.com/open-telemetry/opamp-go/server/types"
+)
+
+type CallbacksStruct struct {
+	OnConnectingFunc      func(request *http.Request) types.ConnectionResponse
+	OnConnectedFunc       func(conn types.Connection)
+	OnMessageFunc         func(conn types.Connection, message *protobufs.AgentToServer)
+	OnDisconnectFunc      func(conn types.Connection, instanceUid string)
+	OnConnectionCloseFunc func(conn types.Connection)
+}
+
+var _ types.Callbacks = (*CallbacksStruct)(nil)
+
+func (c CallbacksStruct) OnConnecting(request *http.Request) types.ConnectionResponse {
+	if c.OnConnectingFunc != nil {
+		return c.OnConnectingFunc(request)
+	}
+	return types.ConnectionResponse{}
+}
+
+func (c CallbacksStruct) OnConnected(conn types.Connection) {
+	if c.OnConnectedFunc != nil {
+		c.OnConnectedFunc(conn)
+	}
+}
+
+func (c CallbacksStruct) OnMessage(conn types.Connection, message *protobufs.AgentToServer) {
+	if c.OnMessageFunc != nil {
+		c.OnMessageFunc(conn, message)
+	}
+}
+
+func (c CallbacksStruct) OnDisconnect(conn types.Connection, instanceUid string) {
+	if c.OnDisconnectFunc != nil {
+		c.OnDisconnectFunc(conn, instanceUid)
+	}
+}
+
+func (c CallbacksStruct) OnConnectionClose(conn types.Connection) {
+	if c.OnConnectionCloseFunc != nil {
+		c.OnConnectionCloseFunc(conn)
+	}
+}
diff --git a/server/server.go b/server/server.go
index e04c4fa..b2df655 100644
--- a/server/server.go
+++ b/server/server.go
@@ -19,6 +19,10 @@ type StartSettings struct {
 	// ListenEndpoint specifies the endpoint to listen on, e.g. "127.0.0.1:4320"
 	ListenEndpoint string
 
+	// ListenPath specifies the URL path on which to accept the OpAMP connections
+	// If this is empty string then Start() will use the default "/v1/opamp" path.
+	ListenPath string
+
 	// Server's TLS configuration.
 	TLSConfig *tls.Config
 }
diff --git a/server/serverimpl.go b/server/serverimpl.go
new file mode 100644
index 0000000..0caf9ea
--- /dev/null
+++ b/server/serverimpl.go
@@ -0,0 +1,178 @@
+package server
+
+import (
+	"context"
+	"errors"
+	"net/http"
+
+	"github.com/gorilla/websocket"
+	"google.golang.org/protobuf/proto"
+
+	"github.com/open-telemetry/opamp-go/client/types"
+	"github.com/open-telemetry/opamp-go/internal"
+	"github.com/open-telemetry/opamp-go/internal/protobufs"
+)
+
+var (
+	errAlreadyStarted = errors.New("already started")
+)
+
+const defaultOpAMPPath = "/v1/opamp"
+
+type server struct {
+	logger   types.Logger
+	settings Settings
+
+	// Upgrader to use to upgrade HTTP to WebSocket.
+	wsUpgrader websocket.Upgrader
+
+	// The listening HTTP Server after successful Start() call. Nil if Start()
+	// is not called or was not successful.
+	httpServer *http.Server
+}
+
+var _ OpAMPServer = (*server)(nil)
+
+func New(logger types.Logger) *server {
+	if logger == nil {
+		logger = &internal.NopLogger{}
+	}
+
+	return &server{logger: logger}
+}
+
+func (s *server) Attach(settings Settings) (HTTPHandlerFunc, error) {
+	s.settings = settings
+	s.wsUpgrader = websocket.Upgrader{}
+	return s.httpHandler, nil
+}
+
+func (s *server) Start(settings StartSettings) error {
+	if s.httpServer != nil {
+		return errAlreadyStarted
+	}
+
+	_, err := s.Attach(settings.Settings)
+	if err != nil {
+		return err
+	}
+
+	// Prepare handling OpAMP incoming HTTP requests on the requests URL path.
+	mux := http.NewServeMux()
+
+	path := settings.ListenPath
+	if path == "" {
+		path = defaultOpAMPPath
+	}
+
+	mux.HandleFunc(path, s.httpHandler)
+
+	hs := &http.Server{
+		Handler: mux,
+		Addr:    settings.ListenEndpoint,
+	}
+	s.httpServer = hs
+
+	// Start the HTTP server in background and return immediately.
+	// TODO: wait until server starts listening or remove the requirement to wait for that.
+	if settings.TLSConfig != nil {
+		hs.TLSConfig = settings.TLSConfig
+		go s.runServer(func() error { return hs.ListenAndServeTLS("", "") })
+	} else {
+		go s.runServer(func() error { return hs.ListenAndServe() })
+	}
+	return nil
+}
+
+func (s *server) runServer(runner func() error) {
+	err := runner()
+	// ErrServerClosed is expected after successful Stop(), so we won't log that
+	// particular error.
+	if err != nil && err != http.ErrServerClosed {
+		s.logger.Errorf("Error running HTTP Server: %v", err)
+	}
+}
+
+func (s *server) Stop(ctx context.Context) error {
+	if s.httpServer != nil {
+		defer func() { s.httpServer = nil }()
+		// This stops accepting new connections. TODO: close existing
+		// connections and wait them to be terminated.
+		return s.httpServer.Shutdown(ctx)
+	}
+	return nil
+}
+
+func (s *server) httpHandler(w http.ResponseWriter, req *http.Request) {
+	if s.settings.Callbacks != nil {
+		resp := s.settings.Callbacks.OnConnecting(req)
+		if !resp.Accept {
+			// HTTP connection is not accepted. Set the response headers.
+			for k, v := range resp.HTTPResponseHeader {
+				w.Header().Set(k, v)
+			}
+			// And write the response status code.
+			w.WriteHeader(resp.HTTPStatusCode)
+			return
+		}
+	}
+
+	// HTTP connection is accepted. Upgrade it to WebSocket.
+	conn, err := s.wsUpgrader.Upgrade(w, req, nil)
+	if err != nil {
+		s.logger.Errorf("Cannot upgrade HTTP connection to WebSocket: %v", err)
+		return
+	}
+
+	// Return from this func to reduce memory usage.
+	// Handle the connection on a separate gorountine.
+	go s.handleWSConnection(conn)
+}
+
+func (s *server) handleWSConnection(wsConn *websocket.Conn) {
+	agentConn := connection{wsConn: wsConn}
+
+	defer func() {
+		// Close the connection when all is done.
+		defer wsConn.Close()
+
+		if s.settings.Callbacks != nil {
+			s.settings.Callbacks.OnConnectionClose(agentConn)
+		}
+	}()
+
+	if s.settings.Callbacks != nil {
+		s.settings.Callbacks.OnConnected(agentConn)
+	}
+
+	// Loop until fail to read from the WebSocket connection.
+	for {
+		// Block until the next message can be read.
+		mt, bytes, err := wsConn.ReadMessage()
+		if err != nil {
+			if !websocket.IsUnexpectedCloseError(err) {
+				s.logger.Errorf("Cannot read a message from WebSocket: %v", err)
+				break
+			}
+			// This is a normal closing of the WebSocket connection.
+			s.logger.Debugf("Agent disconnected: %v", err)
+			break
+		}
+		if mt != websocket.BinaryMessage {
+			s.logger.Errorf("Received unexpected message type from WebSocket: %v", mt)
+			continue
+		}
+
+		// Decode WebSocket message as a Protobuf message.
+		var request protobufs.AgentToServer
+		err = proto.Unmarshal(bytes, &request)
+		if err != nil {
+			s.logger.Errorf("Cannot decode message from WebSocket: %v", err)
+			continue
+		}
+
+		if s.settings.Callbacks != nil {
+			s.settings.Callbacks.OnMessage(agentConn, &request)
+		}
+	}
+}


Test patch:
diff --git a/server/serverimpl_test.go b/server/serverimpl_test.go
new file mode 100644
index 0000000..1d068cb
--- /dev/null
+++ b/server/serverimpl_test.go
@@ -0,0 +1,231 @@
+package server
+
+import (
+	"context"
+	"net/http"
+	"net/http/httptest"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/gorilla/websocket"
+	sharedinternal "github.com/open-telemetry/opamp-go/internal"
+	"github.com/open-telemetry/opamp-go/internal/protobufs"
+	"github.com/open-telemetry/opamp-go/internal/testhelpers"
+	"github.com/open-telemetry/opamp-go/server/internal"
+	"github.com/open-telemetry/opamp-go/server/types"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"google.golang.org/protobuf/proto"
+)
+
+func startServer(t *testing.T, settings *StartSettings) *server {
+	srv := New(&sharedinternal.NopLogger{})
+	require.NotNil(t, srv)
+	if settings.ListenEndpoint == "" {
+		// Find an avaiable port to listne on.
+		settings.ListenEndpoint = testhelpers.GetAvailableLocalAddress()
+	}
+	if settings.ListenPath == "" {
+		settings.ListenPath = "/"
+	}
+	err := srv.Start(*settings)
+	require.NoError(t, err)
+
+	// Wait until the server strts accepting connections.
+	testhelpers.WaitForEndpoint(settings.ListenEndpoint)
+	return srv
+}
+
+func dialClient(serverSettings *StartSettings) (*websocket.Conn, *http.Response, error) {
+	srvUrl := "ws://" + serverSettings.ListenEndpoint + serverSettings.ListenPath
+	return websocket.DefaultDialer.Dial(srvUrl, nil)
+}
+
+func TestServerStartStop(t *testing.T) {
+	srv := startServer(t, &StartSettings{})
+
+	err := srv.Start(StartSettings{})
+	assert.ErrorIs(t, err, errAlreadyStarted)
+
+	err = srv.Stop(context.Background())
+	assert.NoError(t, err)
+}
+
+func TestServerStartRejectConnection(t *testing.T) {
+	callbacks := internal.CallbacksStruct{
+		OnConnectingFunc: func(request *http.Request) types.ConnectionResponse {
+			// Reject the incoming HTTP connection.
+			return types.ConnectionResponse{
+				Accept:             false,
+				HTTPStatusCode:     503,
+				HTTPResponseHeader: map[string]string{"Retry-After": "30"},
+			}
+		},
+	}
+
+	// Start a server.
+	settings := &StartSettings{Settings: Settings{Callbacks: callbacks}}
+	srv := startServer(t, settings)
+	defer srv.Stop(context.Background())
+
+	// Try to connect to the server.
+	conn, resp, err := dialClient(settings)
+
+	// Verify that the connection is rejected and rejection data is available to the client.
+	assert.Nil(t, conn)
+	assert.Error(t, err)
+	assert.EqualValues(t, 503, resp.StatusCode)
+	assert.EqualValues(t, "30", resp.Header.Get("Retry-After"))
+}
+
+func eventually(t *testing.T, f func() bool) {
+	assert.Eventually(t, f, 5*time.Second, 10*time.Millisecond)
+}
+
+func TestServerStartAcceptConnection(t *testing.T) {
+	connectedCalled := int32(0)
+	connectionCloseCalled := int32(0)
+	var srvConn types.Connection
+	callbacks := internal.CallbacksStruct{
+		OnConnectingFunc: func(request *http.Request) types.ConnectionResponse {
+			return types.ConnectionResponse{Accept: true}
+		},
+		OnConnectedFunc: func(conn types.Connection) {
+			atomic.StoreInt32(&connectedCalled, 1)
+			srvConn = conn
+		},
+		OnConnectionCloseFunc: func(conn types.Connection) {
+			atomic.StoreInt32(&connectionCloseCalled, 1)
+			assert.EqualValues(t, srvConn, conn)
+		},
+	}
+
+	// Start a server.
+	settings := &StartSettings{Settings: Settings{Callbacks: callbacks}}
+	srv := startServer(t, settings)
+	defer srv.Stop(context.Background())
+
+	// Connect to the server.
+	conn, resp, err := dialClient(settings)
+
+	// Verify that the connection is successful.
+	assert.NoError(t, err)
+	assert.NotNil(t, conn)
+	assert.EqualValues(t, 101, resp.StatusCode)
+	eventually(t, func() bool { return atomic.LoadInt32(&connectedCalled) == 1 })
+	assert.True(t, atomic.LoadInt32(&connectionCloseCalled) == 0)
+
+	// Close the connection from client side.
+	conn.Close()
+
+	// Verify that the server receives the closing notification.
+	eventually(t, func() bool { return atomic.LoadInt32(&connectionCloseCalled) == 1 })
+}
+
+func TestServerReceiveSendMessage(t *testing.T) {
+	var rcvMsg atomic.Value
+	callbacks := internal.CallbacksStruct{
+		OnConnectingFunc: func(request *http.Request) types.ConnectionResponse {
+			return types.ConnectionResponse{Accept: true}
+		},
+		OnMessageFunc: func(conn types.Connection, message *protobufs.AgentToServer) {
+			// Remember received message.
+			rcvMsg.Store(message)
+
+			// Send a response.
+			response := protobufs.ServerToAgent{
+				InstanceUid:  message.InstanceUid,
+				Capabilities: protobufs.ServerCapabilities_AcceptsStatus,
+			}
+			err := conn.Send(context.Background(), &response)
+			assert.NoError(t, err)
+		},
+	}
+
+	// Start a server.
+	settings := &StartSettings{Settings: Settings{Callbacks: callbacks}}
+	srv := startServer(t, settings)
+	defer srv.Stop(context.Background())
+
+	// Connect using a WebSocket client.
+	conn, _, _ := dialClient(settings)
+	defer conn.Close()
+
+	// Send a message to the server.
+	sendMsg := protobufs.AgentToServer{
+		InstanceUid: "12345678",
+	}
+	bytes, err := proto.Marshal(&sendMsg)
+	require.NoError(t, err)
+	err = conn.WriteMessage(websocket.BinaryMessage, bytes)
+	require.NoError(t, err)
+
+	// Wait until server receives the message.
+	eventually(t, func() bool { return rcvMsg.Load() != nil })
+	assert.True(t, proto.Equal(rcvMsg.Load().(proto.Message), &sendMsg))
+
+	// Read server's response.
+	mt, bytes, err := conn.ReadMessage()
+	require.NoError(t, err)
+	require.EqualValues(t, websocket.BinaryMessage, mt)
+
+	// Decode the response.
+	var response protobufs.ServerToAgent
+	err = proto.Unmarshal(bytes, &response)
+	require.NoError(t, err)
+
+	// Verify the response.
+	assert.EqualValues(t, sendMsg.InstanceUid, response.InstanceUid)
+	assert.EqualValues(t, protobufs.ServerCapabilities_AcceptsStatus, response.Capabilities)
+}
+
+func TestServerAttachAcceptConnection(t *testing.T) {
+	connectedCalled := int32(0)
+	connectionCloseCalled := int32(0)
+	var srvConn types.Connection
+	callbacks := internal.CallbacksStruct{
+		OnConnectingFunc: func(request *http.Request) types.ConnectionResponse {
+			return types.ConnectionResponse{Accept: true}
+		},
+		OnConnectedFunc: func(conn types.Connection) {
+			atomic.StoreInt32(&connectedCalled, 1)
+			srvConn = conn
+		},
+		OnConnectionCloseFunc: func(conn types.Connection) {
+			atomic.StoreInt32(&connectionCloseCalled, 1)
+			assert.EqualValues(t, srvConn, conn)
+		},
+	}
+
+	// Prepare to attach OpAMP server to an HTTP Server created separately.
+	settings := Settings{Callbacks: callbacks}
+	srv := New(&sharedinternal.NopLogger{})
+	require.NotNil(t, srv)
+	handlerFunc, err := srv.Attach(settings)
+	require.NoError(t, err)
+
+	// Create an HTTP server and make it handle OpAMP connections.
+	mux := http.NewServeMux()
+	path := "/opamppath"
+	mux.HandleFunc(path, handlerFunc)
+	hs := httptest.NewServer(mux)
+	defer hs.Close()
+
+	// Wait until server is ready and accepts connections.
+	testhelpers.WaitForEndpoint(hs.Listener.Addr().String())
+
+	// Connect using WebSocket client.
+	srvUrl := "ws://" + hs.Listener.Addr().String() + path
+	conn, resp, err := websocket.DefaultDialer.Dial(srvUrl, nil)
+
+	// Verify that connection is successful.
+	assert.NoError(t, err)
+	assert.NotNil(t, conn)
+	assert.EqualValues(t, 101, resp.StatusCode)
+	eventually(t, func() bool { return atomic.LoadInt32(&connectedCalled) == 1 })
+	assert.True(t, atomic.LoadInt32(&connectionCloseCalled) == 0)
+
+	conn.Close()
+	eventually(t, func() bool { return atomic.LoadInt32(&connectionCloseCalled) == 1 })
+}