Merge pull request #1248 from gohornet/develop

Release 1.1.1

CHANGELOG.md

@@ -2,6 +2,28 @@
 
 All notable changes to this project will be documented in this file.
 
+## [1.1.1] - 22.12.2021
+
+### Changed
+    - Added ulimit and stop_grace_period to docker-compose.yml and documentation (#1242)
+
+### Fixed
+    - Fixed WebSocket disconnecting on Safari browsers (#1243)
+    - Fixed MQTT memory leak (#1246)
+
+### Config file changes
+
+`config.json`
+```diff
+    "prometheus": {
+       ...
+       "coordinatorMetrics": true,
++       "mqttBrokerMetrics": true,
+       "debugMetrics": false,
+       ...
+    },
+```
+
 ## [1.1.0] - 10.12.2021
 
 ### Added

README.md

@@ -36,7 +36,7 @@ _Table of contents_
 
 ## Documentation
 
-Hornet documentation can be found here: https://hornet.docs.iota.org/
+Hornet documentation can be found here: https://wiki.iota.org/hornet/welcome
 
 ## Contributing
 

config.json

@@ -252,6 +252,7 @@
     "restAPIMetrics": true,
     "migrationMetrics": true,
     "coordinatorMetrics": true,
+    "mqttBrokerMetrics": true,
     "debugMetrics": false,
     "goMetrics": false,
     "processMetrics": false,

config_comnet.json

@@ -258,6 +258,7 @@
     "restAPIMetrics": true,
     "migrationMetrics": true,
     "coordinatorMetrics": true,
+    "mqttBrokerMetrics": true,
     "debugMetrics": false,
     "goMetrics": false,
     "processMetrics": false,

config_devnet.json

@@ -258,6 +258,7 @@
     "restAPIMetrics": true,
     "migrationMetrics": true,
     "coordinatorMetrics": true,
+    "mqttBrokerMetrics": true,
     "debugMetrics": false,
     "goMetrics": false,
     "processMetrics": false,

core/app/app.go

@@ -20,7 +20,7 @@ var (
 	Name = "HORNET"
 
 	// Version of the app.
-	Version = "1.1.0"
+	Version = "1.1.1"
 )
 
 var (

docker-compose.yml

@@ -9,6 +9,7 @@ services:
       nofile:
         soft: 8192
         hard: 8192
+    stop_grace_period: 5m
     # Best performance via host network:
     network_mode: host
     # Else:

documentation/docs/api_reference.md

@@ -12,7 +12,7 @@ image: /img/logo/HornetLogo.png
 
 # API Reference
 
-You can find specifications for the REST API for the IOTA node software in the [IOTA REST API reference](https://editor.swagger.io/?url=https://raw.githubusercontent.com/rufsam/protocol-rfcs/master/text/0026-rest-api/rest-api.yaml)
+You can find specifications for the REST API for the IOTA node software in the [IOTA REST API reference](https://editor.swagger.io/?url=https://raw.githubusercontent.com/rufsam/protocol-rfcs/master/text/0026-rest-api/0026-rest-api.yaml)
 
 
-The node event API is in charge of publishing information about events within the node software.  You can find more information in the [Node event API reference](https://playground.asyncapi.io/?load=https://raw.githubusercontent.com/luca-moser/protocol-rfcs/rfc/node-event-api/text/0033-node-event-api/0033-node-event-api.yml) 
+The node event API is in charge of publishing information about events within the node software.  You can find more information in the [Node event API reference](https://studio.asyncapi.com/?url=https://raw.githubusercontent.com/luca-moser/protocol-rfcs/rfc/node-event-api/text/0033-node-event-api/0033-node-event-api.yml) 

documentation/docs/getting_started/security_101.md

@@ -25,7 +25,7 @@ You should consider doing the following before running a node on your device:
 * [Blocking unnecessary ports](#blocking-unnecessary-ports).
 
 ### Securing SSH logins
-If you log into your device through SSH, you should take measures to protect it from unauthorized access. Many guides have been written about this subject. For more information, see [10 Steps to Secure Open SSH](https://blog.devolutions.net/2017/4/10-steps-to-secure-open-ssh). In addition to that, you can also leverage tools such as [Fail2ban](https://www.fail2ban.org/wiki/index.php/Main_Page) to further tighten your nodes security.
+If you log into your device through SSH, you should take measures to protect it from unauthorized access. Many guides have been written about this subject. For more information, see [10 Steps to Secure Open SSH](https://blog.devolutions.net/2017/04/10-steps-to-secure-open-ssh). In addition to that, you can also leverage tools such as [Fail2ban](https://www.fail2ban.org/wiki/index.php/Main_Page) to further tighten your nodes security.
 
 ### Blocking Unnecessary Ports
 Attackers can abuse any open ports on your device. To secure your device against attacks on unused open ports, you should close all ports except those that are in use.

documentation/docs/getting_started/using_docker.md

@@ -81,6 +81,7 @@ docker run \
   --restart always \
   --name hornet\
   --net=host \
+  --ulimit nofile=8192:8192 \
   -d \
   hornet:latest
 ```
@@ -95,10 +96,11 @@ docker run \
 * `--restart always` Instructs Docker to restart the container after Docker reboots.
 * `--name hornet` Name of the running container instance. You can refer to the given container by this name.
 * `--net=host` Instructs Docker to use the host's network, so the network is not isolated. We recommend that you run on host network for better performance.  This way, the container will also open any ports it needs on the host network, so you will not need to specify any ports.
+* `--ulimit nofile=8192:8192` increases the ulimits inside the container. This is important when running with large databases.
 * `-d` Instructs Docker to run the container instance in a detached mode (daemon).
 
 
-You can run `docker stop -t 200 hornet` to gracefully end the process.
+You can run `docker stop -t 300 hornet` to gracefully end the process.
 
 ## Create Username and Password for the Hornet Dashboard
 
@@ -166,20 +168,20 @@ docker start hornet
 You can restart an existing Hornet container by running:
 
 ```bash
-docker restart -t 200 hornet
+docker restart -t 300 hornet
 ```
 
-* `-t 200` Instructs Docker to wait for a grace period before shutting down.
+* `-t 300` Instructs Docker to wait for a grace period before shutting down.
 
 ### Stopping Hornet
 
 You can stop an existing Hornet container by running:
 
 ```bash
-docker stop -t 200 hornet
+docker stop -t 300 hornet
 ```
 
-* `-t 200` Instructs Docker to wait for a grace period before shutting down.
+* `-t 300` Instructs Docker to wait for a grace period before shutting down.
 
 ### Displaying Log Output
 

documentation/docs/getting_started/using_docker_compose.md

@@ -42,6 +42,11 @@ services:
     image: gohornet/hornet:latest
     network_mode: host
     restart: always
+    ulimits:
+      nofile:
+        soft: 8192
+        hard: 8192
+    stop_grace_period: 5m
     cap_drop:
       - ALL
     volumes:
@@ -80,5 +85,5 @@ You can add `-d` to run detached.
 To gracefully stop the container, you can run the following command:
 
 ```sh
-docker-compose down -t 200
+docker-compose down
 ```

documentation/docs/post_installation/configuration.md

@@ -29,21 +29,21 @@ hornet -h --full
 
 ## 1. REST API
 
-| Name                       | Description                                                                                      | Type             |
-| :------------------------- | :----------------------------------------------------------------------------------------------- | :--------------- |
-| bindAddress                | The bind address on which the REST API listens on                                                | string           |
-| [jwtAuth](#jwt-auth)       | Config for JWT auth                                                                              | object           |
-| publicRoutes               | the HTTP REST routes which can be called without authorization. Wildcards using * are allowed.   | array of strings |
-| protectedRoutes            | the HTTP REST routes which need to be called with authorization. Wildcards using * are allowed.  | array of strings |
-| powEnabled                 | Whether the node does PoW if messages are received via API                                       | bool             |
-| powWorkerCount             | The amount of workers used for calculating PoW when issuing messages via API                     | integer          |
-| [limits](#limits)          | Configuration for api limits                                                                     | object           |
+| Name                 | Description                                                                                     | Type             |
+| :------------------- | :---------------------------------------------------------------------------------------------- | :--------------- |
+| bindAddress          | The bind address on which the REST API listens on                                               | string           |
+| [jwtAuth](#jwt-auth) | Config for JWT auth                                                                             | object           |
+| publicRoutes         | the HTTP REST routes which can be called without authorization. Wildcards using * are allowed.  | array of strings |
+| protectedRoutes      | the HTTP REST routes which need to be called with authorization. Wildcards using * are allowed. | array of strings |
+| powEnabled           | Whether the node does PoW if messages are received via API                                      | bool             |
+| powWorkerCount       | The amount of workers used for calculating PoW when issuing messages via API                    | integer          |
+| [limits](#limits)    | Configuration for api limits                                                                    | object           |
 
 ### JWT Auth
 
-| Name    | Description                                                                                                                             | Type   |
-| :------ | :-------------------------------------------------------------------------------------------------------------------------------------- | :----- |
-| salt    | Salt used inside the JWT tokens for the REST API. Change this to a different value to invalidate JWT tokens not matching this new value | string |
+| Name | Description                                                                                                                             | Type   |
+| :--- | :-------------------------------------------------------------------------------------------------------------------------------------- | :----- |
+| salt | Salt used inside the JWT tokens for the REST API. Change this to a different value to invalidate JWT tokens not matching this new value | string |
 
 
 ### Limits
@@ -795,6 +795,7 @@ Example:
 | restAPIMetrics                                | Include restAPI metrics                                      | bool   |
 | migrationMetrics                              | Include migration metrics                                    | bool   |
 | coordinatorMetrics                            | Include coordinator metrics                                  | bool   |
+| mqttBrokerMetrics                             | Include MQTT broker metrics                                  | bool   |
 | debugMetrics                                  | Include debug metrics                                        | bool   |
 | goMetrics                                     | Include go metrics                                           | bool   |
 | processMetrics                                | Include process metrics                                      | bool   |
@@ -825,6 +826,7 @@ Example:
     "restAPIMetrics": true,
     "migrationMetrics": true,
     "coordinatorMetrics": true,
+    "mqttBrokerMetrics": true,
     "debugMetrics": false,
     "goMetrics": false,
     "processMetrics": false,

pkg/model/mselection/heaviest_test.go

@@ -146,7 +146,7 @@ func TestHeaviestSelector_SelectTipsChains(t *testing.T) {
 	assert.Len(t, hps.trackedMessages, 0)
 }
 
-func TestHeaviestSelector_SelectTipsCheckTresholds(t *testing.T) {
+func TestHeaviestSelector_SelectTipsCheckThresholds(t *testing.T) {
 	te, hps := initTest(t)
 	defer te.CleanupTestEnvironment(true)
 

pkg/mqtt/broker.go

@@ -16,7 +16,7 @@ type Broker struct {
 }
 
 // NewBroker creates a new broker.
-func NewBroker(bindAddress string, wsPort int, wsPath string, workerCount int, onSubscribe OnSubscribeHandler, onUnsubscribe OnUnsubscribeHandler) (*Broker, error) {
+func NewBroker(bindAddress string, wsPort int, wsPath string, workerCount int, onSubscribe OnSubscribeHandler, onUnsubscribe OnUnsubscribeHandler, cleanupThreshold int) (*Broker, error) {
 
 	host, port, err := net.SplitHostPort(bindAddress)
 	if err != nil {
@@ -36,7 +36,7 @@ func NewBroker(bindAddress string, wsPort int, wsPath string, workerCount int, o
 		return nil, fmt.Errorf("configure broker config error: %w", err)
 	}
 
-	t := newTopicManager(onSubscribe, onUnsubscribe)
+	t := newTopicManager(onSubscribe, onUnsubscribe, cleanupThreshold)
 
 	b, err := broker.NewBroker(c)
 	if err != nil {
@@ -74,3 +74,8 @@ func (b *Broker) Send(topic string, payload []byte) {
 
 	b.broker.PublishMessage(packet)
 }
+
+// TopicsManagerSize returns the size of the underlying map of the topics manager.
+func (b *Broker) TopicsManagerSize() int {
+	return b.topicManager.Size()
+}

pkg/mqtt/topic_manager.go

@@ -15,8 +15,11 @@ type OnUnsubscribeHandler func(topic []byte)
 type topicManager struct {
 	mem topics.TopicsProvider
 
-	subscribedTopics     map[string]int
-	subscribedTopicsLock sync.RWMutex
+	subscribedTopics        map[string]int
+	subscribedTopicsLock    sync.RWMutex
+	subscribedTopicsDeleted int
+
+	cleanupThreshold int
 
 	onSubscribe   OnSubscribeHandler
 	onUnsubscribe OnUnsubscribeHandler
@@ -49,13 +52,13 @@ func (t *topicManager) Unsubscribe(topic []byte, subscriber interface{}) error {
 
 	err := t.mem.Unsubscribe(topic, subscriber)
 
-	//Ignore error here, always unsubscribe to be safe
+	// ignore error here, always unsubscribe to be safe
 
 	topicName := string(topic)
 	count, has := t.subscribedTopics[topicName]
 	if has {
-		if count <= 0 {
-			delete(t.subscribedTopics, topicName)
+		if count <= 1 {
+			t.deleteTopic(topicName)
 		} else {
 			t.subscribedTopics[topicName] = count - 1
 		}
@@ -82,6 +85,14 @@ func (t *topicManager) Close() error {
 	return t.mem.Close()
 }
 
+// Size returns the size of the underlying map of the topics manager.
+func (t *topicManager) Size() int {
+	t.subscribedTopicsLock.RLock()
+	defer t.subscribedTopicsLock.RUnlock()
+
+	return len(t.subscribedTopics)
+}
+
 func (t *topicManager) hasSubscribers(topicName string) bool {
 	t.subscribedTopicsLock.RLock()
 	defer t.subscribedTopicsLock.RUnlock()
@@ -90,13 +101,35 @@ func (t *topicManager) hasSubscribers(topicName string) bool {
 	return has && count > 0
 }
 
-func newTopicManager(onSubscribe OnSubscribeHandler, onUnsubscribe OnUnsubscribeHandler) *topicManager {
+// cleanupWithoutLocking recreates the subscribedTopics map to release memory for the garbage collector.
+func (t *topicManager) cleanupWithoutLocking() {
+	subscribedTopics := make(map[string]int)
+	for topicName, count := range t.subscribedTopics {
+		subscribedTopics[topicName] = count
+	}
+	t.subscribedTopics = subscribedTopics
+	t.subscribedTopicsDeleted = 0
+}
+
+// deleteTopic deletes a topic from the manager.
+func (t *topicManager) deleteTopic(topicName string) {
+	delete(t.subscribedTopics, topicName)
+
+	// increase the deletion counter to trigger garbage collection
+	t.subscribedTopicsDeleted++
+	if t.cleanupThreshold != 0 && t.subscribedTopicsDeleted >= t.cleanupThreshold {
+		t.cleanupWithoutLocking()
+	}
+}
+
+func newTopicManager(onSubscribe OnSubscribeHandler, onUnsubscribe OnUnsubscribeHandler, cleanupThreshold int) *topicManager {
 
 	mgr := &topicManager{
 		mem:              topics.NewMemProvider(),
 		subscribedTopics: make(map[string]int),
 		onSubscribe:      onSubscribe,
 		onUnsubscribe:    onUnsubscribe,
+		cleanupThreshold: cleanupThreshold,
 	}
 
 	// The normal MQTT broker uses the `mem` topic manager internally, so first unregister the default one.

pkg/protocol/gossip/service.go

@@ -593,7 +593,7 @@ func (s *Service) registerLoggerOnEvents() {
 		s.LogInfof("canceled inbound protocol stream from %s: %s", remotePeer, reason)
 	}))
 	s.Events.Error.Attach(events.NewClosure(func(err error) {
-		s.LogError(err)
+		s.LogWarn(err)
 	}))
 }
 

plugins/dashboard/plugin.go

@@ -129,9 +129,12 @@ func configure() {
 	}
 
 	upgrader = &websocket.Upgrader{
-		HandshakeTimeout:  webSocketWriteTimeout,
-		CheckOrigin:       func(r *http.Request) bool { return true }, // allow any origin for websocket connections
-		EnableCompression: true,
+		HandshakeTimeout: webSocketWriteTimeout,
+		CheckOrigin:      func(r *http.Request) bool { return true }, // allow any origin for websocket connections
+		// Disable compression due to incompatibilities with latest Safari browsers:
+		// https://github.com/tilt-dev/tilt/issues/4746
+		// https://github.com/gorilla/websocket/issues/731
+		EnableCompression: false,
 	}
 
 	hub = websockethub.NewHub(Plugin.Logger(), upgrader, broadcastQueueSize, clientSendChannelSize, maxWebsocketMessageSize)

plugins/mqtt/params.go

@@ -7,12 +7,14 @@ import (
 )
 
 const (
-	// the bind address on which the MQTT broker listens on
+	// the bind address on which the MQTT broker listens on.
 	CfgMQTTBindAddress = "mqtt.bindAddress"
-	// the port of the WebSocket MQTT broker
+	// the port of the WebSocket MQTT broker.
 	CfgMQTTWSPort = "mqtt.wsPort"
-	// the number of parallel workers the MQTT broker uses to publish messages
+	// the number of parallel workers the MQTT broker uses to publish messages.
 	CfgMQTTWorkerCount = "mqtt.workerCount"
+	// the number of deleted topics that trigger a garbage collection of the topic manager.
+	CfgMQTTTopicCleanupThreshold = "mqtt.topicCleanupThreshold"
 )
 
 var params = &node.PluginParams{
@@ -22,6 +24,7 @@ var params = &node.PluginParams{
 			fs.String(CfgMQTTBindAddress, "localhost:1883", "bind address on which the MQTT broker listens on")
 			fs.Int(CfgMQTTWSPort, 1888, "port of the WebSocket MQTT broker")
 			fs.Int(CfgMQTTWorkerCount, 100, "number of parallel workers the MQTT broker uses to publish messages")
+			fs.Int(CfgMQTTTopicCleanupThreshold, 10000, "number of deleted topics that trigger a garbage collection of the topic manager")
 			return fs
 		}(),
 	},