Repository: hyperledger/firefly (go)

Issue:
Add a status endpoint to view currently connected websocket clients

Pull request description:
Add /api/v1/status/websockets
Fixes #679

Patch:
diff --git a/docs/swagger/swagger.yaml b/docs/swagger/swagger.yaml
index 852136544..b105a443b 100644
--- a/docs/swagger/swagger.yaml
+++ b/docs/swagger/swagger.yaml
@@ -14101,5 +14101,68 @@ paths:
           description: Success
         default:
           description: ""
+  /status/websockets:
+    get:
+      description: 'TODO: Description'
+      operationId: getStatusWebSockets
+      parameters:
+      - description: Server-side request timeout (millseconds, or set a custom suffix
+          like 10s)
+        in: header
+        name: Request-Timeout
+        schema:
+          default: 120s
+          type: string
+      responses:
+        "200":
+          content:
+            application/json:
+              schema:
+                properties:
+                  connections:
+                    description: List of currently active websocket client connections
+                    items:
+                      description: List of currently active websocket client connections
+                      properties:
+                        id:
+                          description: The unique ID assigned to this client connection
+                          type: string
+                        remoteAddress:
+                          description: The remote address of the connected client
+                            (if available)
+                          type: string
+                        subscriptions:
+                          description: List of subscriptions currently started by
+                            this client
+                          items:
+                            description: List of subscriptions currently started by
+                              this client
+                            properties:
+                              ephemeral:
+                                description: Indicates whether the subscription is
+                                  ephemeral (vs durable)
+                                type: boolean
+                              name:
+                                description: The subscription name (for durable subscriptions
+                                  only)
+                                type: string
+                              namespace:
+                                description: The subscription namespace
+                                type: string
+                            type: object
+                          type: array
+                        userAgent:
+                          description: The user agent of the connected client (if
+                            available)
+                          type: string
+                      type: object
+                    type: array
+                  enabled:
+                    description: Indicates whether the websockets plugin is enabled
+                    type: boolean
+                type: object
+          description: Success
+        default:
+          description: ""
 servers:
 - url: http://localhost:5000
diff --git a/internal/apiserver/route_get_status_websockets.go b/internal/apiserver/route_get_status_websockets.go
new file mode 100644
index 000000000..04d2f6573
--- /dev/null
+++ b/internal/apiserver/route_get_status_websockets.go
@@ -0,0 +1,41 @@
+// Copyright © 2022 Kaleido, Inc.
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package apiserver
+
+import (
+	"net/http"
+
+	"github.com/hyperledger/firefly/internal/coremsgs"
+	"github.com/hyperledger/firefly/internal/oapispec"
+	"github.com/hyperledger/firefly/pkg/fftypes"
+)
+
+var getStatusWebSockets = &oapispec.Route{
+	Name:            "getStatusWebSockets",
+	Path:            "status/websockets",
+	Method:          http.MethodGet,
+	PathParams:      nil,
+	QueryParams:     nil,
+	FilterFactory:   nil,
+	Description:     coremsgs.APIMessageTBD,
+	JSONInputValue:  nil,
+	JSONOutputValue: func() interface{} { return &fftypes.WebSocketStatus{} },
+	JSONOutputCodes: []int{http.StatusOK},
+	JSONHandler: func(r *oapispec.APIRequest) (output interface{}, err error) {
+		return getOr(r.Ctx).Events().GetWebSocketStatus(), nil
+	},
+}
diff --git a/internal/apiserver/routes.go b/internal/apiserver/routes.go
index 400784197..1929f9c9b 100644
--- a/internal/apiserver/routes.go
+++ b/internal/apiserver/routes.go
@@ -66,6 +66,7 @@ var routes = []*oapispec.Route{
 	getStatus,
 	getStatusBatchManager,
 	getStatusPins,
+	getStatusWebSockets,
 	getSubscriptionByID,
 	getSubscriptions,
 	getTokenAccountPools,
diff --git a/internal/coremsgs/en_struct_descriptions.go b/internal/coremsgs/en_struct_descriptions.go
index 07fcd4495..3ac1dc459 100644
--- a/internal/coremsgs/en_struct_descriptions.go
+++ b/internal/coremsgs/en_struct_descriptions.go
@@ -565,4 +565,19 @@ var (
 
 	// ContractCallResponse field descriptions
 	ContractCallResponseID = ffm("ContractCallResponse.id", "The UUID of the operation created to track the transaction submission through to completion")
+
+	// WebSocketStatus field descriptions
+	WebSocketStatusEnabled     = ffm("WebSocketStatus.enabled", "Indicates whether the websockets plugin is enabled")
+	WebSocketStatusConnections = ffm("WebSocketStatus.connections", "List of currently active websocket client connections")
+
+	// WSConnectionStatus field descriptions
+	WSConnectionStatusID            = ffm("WSConnectionStatus.id", "The unique ID assigned to this client connection")
+	WSConnectionStatusRemoteAddress = ffm("WSConnectionStatus.remoteAddress", "The remote address of the connected client (if available)")
+	WSConnectionStatusUserAgent     = ffm("WSConnectionStatus.userAgent", "The user agent of the connected client (if available)")
+	WSConnectionStatusSubscriptions = ffm("WSConnectionStatus.subscriptions", "List of subscriptions currently started by this client")
+
+	// WSSubscriptionStatus field descriptions
+	WSSubscriptionStatusEphemeral = ffm("WSSubscriptionStatus.ephemeral", "Indicates whether the subscription is ephemeral (vs durable)")
+	WSSubscriptionStatusNamespace = ffm("WSSubscriptionStatus.namespace", "The subscription namespace")
+	WSSubscriptionStatusName      = ffm("WSSubscriptionStatus.name", "The subscription name (for durable subscriptions only)")
 )
diff --git a/internal/events/event_manager.go b/internal/events/event_manager.go
index 8531411f7..e971ff3c1 100644
--- a/internal/events/event_manager.go
+++ b/internal/events/event_manager.go
@@ -58,6 +58,7 @@ type EventManager interface {
 	DeletedSubscriptions() chan<- *fftypes.UUID
 	DeleteDurableSubscription(ctx context.Context, subDef *fftypes.Subscription) (err error)
 	CreateUpdateDurableSubscription(ctx context.Context, subDef *fftypes.Subscription, mustNew bool) (err error)
+	GetWebSocketStatus() *fftypes.WebSocketStatus
 	Start() error
 	WaitStop()
 
@@ -256,3 +257,7 @@ func (em *eventManager) GetPlugins() []*fftypes.NodeStatusPlugin {
 
 	return eventsArray
 }
+
+func (em *eventManager) GetWebSocketStatus() *fftypes.WebSocketStatus {
+	return em.subManager.getWebSocketStatus()
+}
diff --git a/internal/events/subscription_manager.go b/internal/events/subscription_manager.go
index ecedca796..25877da7d 100644
--- a/internal/events/subscription_manager.go
+++ b/internal/events/subscription_manager.go
@@ -27,6 +27,7 @@ import (
 	"github.com/hyperledger/firefly/internal/definitions"
 	"github.com/hyperledger/firefly/internal/events/eifactory"
 	"github.com/hyperledger/firefly/internal/events/system"
+	"github.com/hyperledger/firefly/internal/events/websockets"
 	"github.com/hyperledger/firefly/internal/retry"
 	"github.com/hyperledger/firefly/internal/txcommon"
 	"github.com/hyperledger/firefly/pkg/config"
@@ -220,7 +221,7 @@ func (sm *subscriptionManager) newOrUpdatedDurableSubscription(id *fftypes.UUID)
 			return
 		}
 		// Need to close the old one
-		loaded, dispatchers := sm.closeDurabeSubscriptionLocked(subDef.ID)
+		loaded, dispatchers := sm.closeDurableSubscriptionLocked(subDef.ID)
 		if loaded {
 			// Outside the lock, close out the active dispatchers
 			sm.mux.Unlock()
@@ -236,7 +237,7 @@ func (sm *subscriptionManager) newOrUpdatedDurableSubscription(id *fftypes.UUID)
 	}
 }
 
-func (sm *subscriptionManager) closeDurabeSubscriptionLocked(id *fftypes.UUID) (bool, []*eventDispatcher) {
+func (sm *subscriptionManager) closeDurableSubscriptionLocked(id *fftypes.UUID) (bool, []*eventDispatcher) {
 	var dispatchers []*eventDispatcher
 	// Remove it from the list of durable subs (if there)
 	_, loaded := sm.durableSubs[*id]
@@ -256,7 +257,7 @@ func (sm *subscriptionManager) closeDurabeSubscriptionLocked(id *fftypes.UUID) (
 
 func (sm *subscriptionManager) deletedDurableSubscription(id *fftypes.UUID) {
 	sm.mux.Lock()
-	loaded, dispatchers := sm.closeDurabeSubscriptionLocked(id)
+	loaded, dispatchers := sm.closeDurableSubscriptionLocked(id)
 	sm.mux.Unlock()
 
 	log.L(sm.ctx).Infof("Cleaning up subscription %s loaded=%t dispatchers=%d", id, loaded, len(dispatchers))
@@ -565,3 +566,11 @@ func (sm *subscriptionManager) deliveryResponse(ei events.Plugin, connID string,
 	sm.mux.Unlock()
 	dispatcher.deliveryResponse(inflight)
 }
+
+func (sm *subscriptionManager) getWebSocketStatus() *fftypes.WebSocketStatus {
+	pluginName := (*websockets.WebSockets)(nil).Name()
+	if plugin, ok := sm.transports[pluginName]; ok {
+		return plugin.(*websockets.WebSockets).GetStatus()
+	}
+	return &fftypes.WebSocketStatus{Enabled: false}
+}
diff --git a/internal/events/websockets/websocket_connection.go b/internal/events/websockets/websocket_connection.go
index faa34e754..a2a5dd460 100644
--- a/internal/events/websockets/websocket_connection.go
+++ b/internal/events/websockets/websocket_connection.go
@@ -50,9 +50,11 @@ type websocketConnection struct {
 	inflight     []*fftypes.EventDeliveryResponse
 	mux          sync.Mutex
 	closed       bool
+	remoteAddr   string
+	userAgent    string
 }
 
-func newConnection(pCtx context.Context, ws *WebSockets, wsConn *websocket.Conn) *websocketConnection {
+func newConnection(pCtx context.Context, ws *WebSockets, wsConn *websocket.Conn, req *http.Request) *websocketConnection {
 	connID := fftypes.NewUUID().String()
 	ctx := log.WithLogField(pCtx, "websocket", connID)
 	ctx, cancelCtx := context.WithCancel(ctx)
@@ -65,6 +67,8 @@ func newConnection(pCtx context.Context, ws *WebSockets, wsConn *websocket.Conn)
 		sendMessages: make(chan interface{}),
 		senderDone:   make(chan struct{}),
 		receiverDone: make(chan struct{}),
+		remoteAddr:   req.RemoteAddr,
+		userAgent:    req.UserAgent(),
 	}
 	go wc.sendLoop()
 	go wc.receiveLoop()
diff --git a/internal/events/websockets/websockets.go b/internal/events/websockets/websockets.go
index c66384bcb..4947c9053 100644
--- a/internal/events/websockets/websockets.go
+++ b/internal/events/websockets/websockets.go
@@ -95,7 +95,7 @@ func (ws *WebSockets) ServeHTTP(res http.ResponseWriter, req *http.Request) {
 	}
 
 	ws.connMux.Lock()
-	wc := newConnection(ws.ctx, ws, wsConn)
+	wc := newConnection(ws.ctx, ws, wsConn, req)
 	ws.connections[wc.connID] = wc
 	ws.connMux.Unlock()
 
@@ -138,3 +138,38 @@ func (ws *WebSockets) WaitClosed() {
 		ws.waitClose()
 	}
 }
+
+func (ws *WebSockets) GetStatus() *fftypes.WebSocketStatus {
+	status := &fftypes.WebSocketStatus{
+		Enabled:     true,
+		Connections: make([]*fftypes.WSConnectionStatus, 0),
+	}
+
+	ws.connMux.Lock()
+	connections := make([]*websocketConnection, 0, len(ws.connections))
+	for _, c := range ws.connections {
+		connections = append(connections, c)
+	}
+	ws.connMux.Unlock()
+
+	for _, wc := range connections {
+		wc.mux.Lock()
+		conn := &fftypes.WSConnectionStatus{
+			ID:            wc.connID,
+			RemoteAddress: wc.remoteAddr,
+			UserAgent:     wc.userAgent,
+			Subscriptions: make([]*fftypes.WSSubscriptionStatus, 0),
+		}
+		status.Connections = append(status.Connections, conn)
+		for _, s := range wc.started {
+			sub := &fftypes.WSSubscriptionStatus{
+				Name:      s.name,
+				Namespace: s.namespace,
+				Ephemeral: s.ephemeral,
+			}
+			conn.Subscriptions = append(conn.Subscriptions, sub)
+		}
+		wc.mux.Unlock()
+	}
+	return status
+}
diff --git a/mocks/eventmocks/event_manager.go b/mocks/eventmocks/event_manager.go
index beea61b9a..db893befa 100644
--- a/mocks/eventmocks/event_manager.go
+++ b/mocks/eventmocks/event_manager.go
@@ -132,6 +132,22 @@ func (_m *EventManager) GetPlugins() []*fftypes.NodeStatusPlugin {
 	return r0
 }
 
+// GetWebSocketStatus provides a mock function with given fields:
+func (_m *EventManager) GetWebSocketStatus() *fftypes.WebSocketStatus {
+	ret := _m.Called()
+
+	var r0 *fftypes.WebSocketStatus
+	if rf, ok := ret.Get(0).(func() *fftypes.WebSocketStatus); ok {
+		r0 = rf()
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).(*fftypes.WebSocketStatus)
+		}
+	}
+
+	return r0
+}
+
 // NewEvents provides a mock function with given fields:
 func (_m *EventManager) NewEvents() chan<- int64 {
 	ret := _m.Called()
diff --git a/pkg/fftypes/websocket_status.go b/pkg/fftypes/websocket_status.go
new file mode 100644
index 000000000..cae42882e
--- /dev/null
+++ b/pkg/fftypes/websocket_status.go
@@ -0,0 +1,35 @@
+// Copyright © 2022 Kaleido, Inc.
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package fftypes
+
+type WSSubscriptionStatus struct {
+	Ephemeral bool   `ffstruct:"WSSubscriptionStatus" json:"ephemeral"`
+	Namespace string `ffstruct:"WSSubscriptionStatus" json:"namespace"`
+	Name      string `ffstruct:"WSSubscriptionStatus" json:"name,omitempty"`
+}
+
+type WSConnectionStatus struct {
+	ID            string                  `ffstruct:"WSConnectionStatus" json:"id"`
+	RemoteAddress string                  `ffstruct:"WSConnectionStatus" json:"remoteAddress"`
+	UserAgent     string                  `ffstruct:"WSConnectionStatus" json:"userAgent"`
+	Subscriptions []*WSSubscriptionStatus `ffstruct:"WSConnectionStatus" json:"subscriptions"`
+}
+
+type WebSocketStatus struct {
+	Enabled     bool                  `ffstruct:"WebSocketStatus" json:"enabled"`
+	Connections []*WSConnectionStatus `ffstruct:"WebSocketStatus" json:"connections"`
+}


Test patch:
diff --git a/internal/apiserver/route_get_status_websockets_test.go b/internal/apiserver/route_get_status_websockets_test.go
new file mode 100644
index 000000000..6799abbd4
--- /dev/null
+++ b/internal/apiserver/route_get_status_websockets_test.go
@@ -0,0 +1,40 @@
+// Copyright © 2021 Kaleido, Inc.
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package apiserver
+
+import (
+	"net/http/httptest"
+	"testing"
+
+	"github.com/hyperledger/firefly/mocks/eventmocks"
+	"github.com/hyperledger/firefly/pkg/fftypes"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestGetStatusWebSockets(t *testing.T) {
+	o, r := newTestAPIServer()
+	mem := &eventmocks.EventManager{}
+	o.On("Events").Return(mem)
+	req := httptest.NewRequest("GET", "/api/v1/status/websockets", nil)
+	req.Header.Set("Content-Type", "application/json; charset=utf-8")
+	res := httptest.NewRecorder()
+
+	mem.On("GetWebSocketStatus").Return(&fftypes.WebSocketStatus{})
+	r.ServeHTTP(res, req)
+
+	assert.Equal(t, 200, res.Result().StatusCode)
+}
diff --git a/internal/events/event_manager_test.go b/internal/events/event_manager_test.go
index 0bc201f60..2ef8fe39d 100644
--- a/internal/events/event_manager_test.go
+++ b/internal/events/event_manager_test.go
@@ -40,6 +40,7 @@ import (
 	"github.com/hyperledger/firefly/mocks/txcommonmocks"
 	"github.com/hyperledger/firefly/pkg/config"
 	"github.com/hyperledger/firefly/pkg/database"
+	"github.com/hyperledger/firefly/pkg/events"
 	"github.com/hyperledger/firefly/pkg/fftypes"
 	"github.com/sirupsen/logrus"
 	"github.com/stretchr/testify/assert"
@@ -423,3 +424,15 @@ func TestGetPlugins(t *testing.T) {
 
 	assert.ElementsMatch(t, em.GetPlugins(), expectedPlugins)
 }
+
+func TestGetWebSocketStatus(t *testing.T) {
+	em, cancel := newTestEventManager(t)
+	defer cancel()
+
+	status := em.GetWebSocketStatus()
+	assert.Equal(t, true, status.Enabled)
+
+	em.subManager.transports = make(map[string]events.Plugin)
+	status = em.GetWebSocketStatus()
+	assert.Equal(t, false, status.Enabled)
+}
diff --git a/internal/events/websockets/websockets_test.go b/internal/events/websockets/websockets_test.go
index eb2f83219..d9901b80b 100644
--- a/internal/events/websockets/websockets_test.go
+++ b/internal/events/websockets/websockets_test.go
@@ -538,3 +538,30 @@ func TestWebsocketSendAfterClose(t *testing.T) {
 	err = connection.send(map[string]string{"foo": "bar"})
 	assert.Regexp(t, "FF10290", err)
 }
+
+func TestWebsocketStatus(t *testing.T) {
+	ws := &WebSockets{
+		ctx:         context.Background(),
+		connections: make(map[string]*websocketConnection),
+	}
+	ws.connections["id1"] = &websocketConnection{
+		connID: "id1",
+		started: []*websocketStartedSub{
+			{ephemeral: false, name: "name1", namespace: "ns1"},
+		},
+		remoteAddr: "otherhost",
+		userAgent:  "user",
+	}
+
+	status := ws.GetStatus()
+
+	assert.Equal(t, true, status.Enabled)
+	assert.Len(t, status.Connections, 1)
+	assert.Equal(t, "id1", status.Connections[0].ID)
+	assert.Equal(t, "otherhost", status.Connections[0].RemoteAddress)
+	assert.Equal(t, "user", status.Connections[0].UserAgent)
+	assert.Len(t, status.Connections[0].Subscriptions, 1)
+	assert.Equal(t, false, status.Connections[0].Subscriptions[0].Ephemeral)
+	assert.Equal(t, "ns1", status.Connections[0].Subscriptions[0].Namespace)
+	assert.Equal(t, "name1", status.Connections[0].Subscriptions[0].Name)
+}