There is a mechanism to detect if a connection somewhere in the
cluster is using the session ID of an existing one, and if so,
close one as a duplicate.
However, when different domains are used, they should not be considered
duplicates.
Signed-off-by: Ivan Kozlovic <ivan@synadia.com>
When there was a failure to delete a QoS1 consumer, the session
would still be deleted, which would cause orphaned consumers.
In case of error, the session record will not be deleted, which means
that it is still possible to restart the session and then close
it (with the clean flag).
Relates to #3116
Signed-off-by: Ivan Kozlovic <ivan@synadia.com>
Ability to override the stream and consumers replica count, which is by default
determined based on the cluster size.
```
mqtt {
port: 1883
stream_replicas: 5
consumer_replicas: 1
}
```
The above would allow *new* MQTT streams to be created with a replicas
factor of 5 (it will be an error if the cluster does not have that
many nodes, and error will occur at runtime when the first client
on a given account connects), and new consumers would be R=1.
The MQTT existing streams/consumers for an account are not modified.
The stream_replicas can also obviously be reduced to 1 for a cluster
of 3 nodes if one desire to have those streams as R=1.
A value of 0 or negative is considered letting the server pick
the value (from 1 to 3 depending on standalone/cluster size).
There is another property that allows the consumers to be created
with memory storage instead of file:
```
mqtt {
..
consumer_memory_storage: true
}
```
Those new settings are global and apply to new streams/consumers
only.
Related to #3116
Signed-off-by: Ivan Kozlovic <ivan@synadia.com>
Update warning
Signed-off-by: Ivan Kozlovic <ivan@synadia.com>
Also had to change all references from `path.` to `filepath.` when
dealing with files, so that it works properly on Windows.
Fixed also lots of tests to defer the shutdown of the server
after the removal of the storage, and fixed some config files
directories to use the single quote `'` to surround the file path,
again to work on Windows.
Signed-off-by: Ivan Kozlovic <ivan@synadia.com>
This allows stream placement to overflow to adjacent clusters.
We also do more balanced placement based on resources (store or mem). We can continue to expand this as well.
We also introduce an account requirement that stream configs contain a MaxBytes value.
We now track account limits and server limits more distinctly, and do not reserver server resources based on account limits themselves.
Signed-off-by: Derek Collison <derek@nats.io>
Along a leaf node connection, unless the system account is shared AND the JetStream domain name is identical, the default JetStream traffic (without a domain set) will be denied.
As a consequence, all clients that wants to access a domain that is not the one in the server they are connected to, a domain name must be specified.
Affected from this change are setups where: a leaf node had no local JetStream OR the server the leaf node connected to had no local JetStream.
One of the two accounts that are connected via a leaf node remote, must have no JetStream enabled.
The side that does not have JetStream enabled, will loose JetStream access and it's clients must set `nats.Domain` manually.
For workarounds on how to restore the old behavior, look at:
https://github.com/nats-io/nats-server/pull/2693#issuecomment-996212582
New config values added:
`default_js_domain` is a mapping from account to domain, settable when JetStream is not enabled in an account.
`extension_hint` are hints for non clustered server to start in clustered mode (and be usable to extend)
`js_domain` is a way to set the JetStream domain to use for mqtt.
Signed-off-by: Matthias Hanel <mh@synadia.com>
This way, any log statement for a client will include the client id,
similar to how the server now logs information about NATS clients
(such as language, version, connection name).
Also adding a debug statement once the client has successfully connected.
Here is how this will look like for a client with client id "client_0".
```
[69591] 2021/10/06 10:06:50.837977 [DBG] [::1]:57415 - mid:18 - Client connection created
[69591] 2021/10/06 10:06:50.839871 [DBG] [::1]:57415 - mid:18 - "client_0" - Client connected
[69591] 2021/10/06 10:07:00.627307 [DBG] [::1]:57415 - mid:18 - "client_0" - Client connection closed: Client Closed
```
All log statements will be affected, for instance here is an auth error:
```
[69591] 2021/10/06 10:09:48.618964 [DBG] [::1]:57424 - mid:23 - Client connection created
[69591] 2021/10/06 10:09:48.619015 [ERR] [::1]:57424 - mid:23 - "client_0" - authentication error - User "mqtt"
[69591] 2021/10/06 10:09:48.619026 [DBG] [::1]:57424 - mid:23 - "client_0" - Client connection closed: Authentication Failure
[69591] 2021/10/06 10:09:48.619038 [ERR] [::1]:57424 - mid:23 - "client_0" - unable to connect: authentication error
```
Resolves#2587
Signed-off-by: Ivan Kozlovic <ivan@synadia.com>
Creating a stream will become idempotent, so assuming that we
should try to transfer the old session streams only on success
will no longer work.
Added a test that checks that "stream" list is queried only once
which means transfer was attempted only once after the second
cluster restart and new connection.
Signed-off-by: Ivan Kozlovic <ivan@synadia.com>
ClientID has been added to various monitoring objects. Also, added
the ability to filter connections on `client_id`.
On auth violation, the proper code was not invoked, which meant
that no disconnect event (with auth reason) would be published.
Resolves#2270
Signed-off-by: Ivan Kozlovic <ivan@synadia.com>
With the availability of a "max message per subject" for a given
stream, it is possible to replace individual streams that were
created per session with a single stream that gets all sessions
as a single message per subject, which subject is composed of
the session client ID hash.
The first time the new stream is created for a given account,
all existing MQTT session streams will be transferred to the
new mux'ed MQTT session stream.
Signed-off-by: Ivan Kozlovic <ivan@synadia.com>
The issue was that the subscription created for the MQTT client
was resulting in creation of a shadow subscription which did not
have the mqtt specific object attached, which would cause the
panic when accessing it in the sub's icb.
After that, it was discovered that the wrong subject was passed
to deliverMsg(), so fixed that too so that the icb callback gets
the proper transformed subject.
Resolves#2265
Signed-off-by: Ivan Kozlovic <ivan@synadia.com>
Say with a cluster of 3, all MQTT assets are created with a replicas
of 3. However, when a server is shutdown, then any new MQTT client
will fail to connect because we try to create a session stream
with R(3), which leads to insufficient resources.
The longer term solution should be for the server to allow the
creation of an asset with a R() value that is bigger than the
current number of running servers as long as there is quorum.
For now, we will reduce the R() value for the sessions if we get
an "insufficient resources" error.
Note that the other assets still will use the compute R() based
on cluster size. So the first time that a client on a given
account is started, we will still need to have R() == cluster size
(at least for R(3)).
Partially resolves#2226
Signed-off-by: Ivan Kozlovic <ivan@synadia.com>
This will solve the issue of naming the durable per server for
the "retained messages" stream in situation where a cluster
of servers would not have JetStream defined but connect to another
cluster that has it. All the servers within the cluster without
JetStream would cause the durable's delivery subject to be updated
to the last server starting the durable.
Updated the check for mqtt requiring JetStream if running in
standalone mode to check that no leafnode configuration is present.
Replaced use of fmt.Errorf() when the string was static with
errors created with errors.New(). Updated tests that were checking
for errors to use those errors instead of repeating the string.
Added test that has a hub cluster with JS enabled and a remote server
that has mqtt{} without JetStream and ensure that this works.
Signed-off-by: Ivan Kozlovic <ivan@synadia.com>
Some issues that have been fixed would manifest by timeouts on
connect, unexpected memory usage on high publish message rate.
Some details:
- Replies were not always GW routed properly because we were looking
at the wrong connection's rsubs
- GW routed replies would not be found because they were tracked
in the subscription's client object, which may not be the same used
to send the reply
- Increased the mqtt timeout to wait for JS replies since in some
tests it was sometimes taking more than the original 2 seconds
- Incoming gateway messages destined for an MQTT internal subscription
may have been rejected as a no interest if the account had service imports
- Don't use time.After(), instead create explicit timer so it can
be stopped when not timing out.
- Unnecessary copy of a slice since we were converting to a string anyway.
Signed-off-by: Ivan Kozlovic <ivan@synadia.com>
In cluster mode, a sub connects to server 1, another on server 2.
A publisher connects to server 2 and publishes a retained message.
If both subs restart they would properly receive the retained message.
However, if the publisher sens an empty message that "removes" the
retained message for this topic, and then consumer that connects to
server 1 restarts, it would not receive the retained message as it
should.
Signed-off-by: Ivan Kozlovic <ivan@synadia.com>
In a setup with shared system account and a cluster of leaf nodes,
the JS requests did not contain the origin cluster, which caused
assets to possibly be created in the HUB. With this change, the
assets will be created in the origin cluster.
Also, removed use of acc.JetStreamEnabled() but instead fail
start of the server if mqtt is enabled in standalone mode and JS
is not enabled. If JS is enabled, we will get proper error if
account has no JS enabled.
Signed-off-by: Ivan Kozlovic <ivan@synadia.com>
Websocket is currently not supported for MQTT clients. When a
client tries to connect with websocket protocol to the MQTT port,
the error message: `mid:9 - not connected` would be logged, which
is not really telling.
The server will now guess if the connection was websocket and report
a more appropriate error message, such as:
```
invalid connection, websocket currently not supported
```
Resolves#2126
Signed-off-by: Ivan Kozlovic <ivan@synadia.com>
Currently, we use ReadyForConnections in server tests to wait for the
server to be ready. However, when this fails we don't get a clue about
why it failed.
This change adds a new unexported method called readyForConnections that
returns an error describing which check failed. The exported
ReadyForConnections version works exactly as before. The unexported
version gets used in internal tests only.
Currently in tests, we have calls to os.Remove and os.RemoveAll where we
don't check the returned error. This hides useful error messages when
tests fail to run, such as "too many open files".
This change checks for more filesystem related errors and calls t.Fatal
if there is an error.
In error conditions or when replacing an existing sub qos1 to qos0,
we were unsubscribing the NATS subscription, but that would not
have been propagated across the cluster.
Also fixed a flapper
Signed-off-by: Ivan Kozlovic <ivan@synadia.com>
- Keep track of published topic to avoid conversion in some case
- Rework tracing of some MQTT protocols
- Rework topic-to-subject conversion
- Update some tests
Signed-off-by: Ivan Kozlovic <ivan@synadia.com>
- Added non-public stream and consumer configuration options to
achieve the "no subject" and "no interest" capabilities. Had
to implement custom FileStreamInfo and FileConsumerInfo marshal/
unmarshal methods so that those non public fields can be
persisted/recovered properly.
- Restored some of JS original code (since now can use config
instead of passing booleans to the functions).
- Use RLock for deliveryFormsCycle() check (unrelated to MQTT).
- Removed restriction on creating streams with MQTT prefix.
- Preventing API deletion of internal streams and their consumers.
- Added comment on Sublist's ReverseMatch method.
Signed-off-by: Ivan Kozlovic <ivan@synadia.com>
MQTT streams are special in that we do not set subjects in the config
since they capture all subjects. Otherwise, we would have been forced
to create a stream on say "MQTT.>" but then all publishes would have
to be prefixed with "MQTT." in order for them to be captured.
However, if one uses the "nats" tool to inspect those streams, the tool
would fail with:
```
server response is not a valid "io.nats.jetstream.api.v1.stream_info_response" message:
(root): Must validate one and only one schema (oneOf)
config: subjects is required
config: Must validate all the schemas (allOf)
```
To solve that, if we detect that user asks for the MQTT streams, we
artificially set the returned config's subject to ">".
Alternatively, we may want to not return those streams at all, although
there may be value to see the info for mqtt streams/consumers.
Signed-off-by: Ivan Kozlovic <ivan@synadia.com>
Tests dealing with MQTT "will" needed to wait for the server to
process the MQTT client close of the connection. Only then we
have the guarantee that the server produced the "will" message.
Signed-off-by: Ivan Kozlovic <ivan@synadia.com>
This PR introduces native support for MQTT clients. It requires use
of accounts with JetStream enabled. Since as of now clustering is
not available, MQTT will be limited to single instance.
Only QoS 0 and 1 are supported at the moment. MQTT clients can
exchange messages with NATS clients and vice-versa.
Since JetStream is required, accounts with JetStream enabled must
exist in order for an MQTT client to connect to the NATS Server.
The administrator can limit the users that can use MQTT with the
allowed_connection_types option in the user section. For instance:
```
accounts {
mqtt {
users [
{user: all, password: pwd, allowed_connection_types: ["STANDARD", "WEBSOCKET", "MQTT"]}
{user: mqtt_only, password: pwd, allowed_connection_types: "MQTT"}
]
jetstream: enabled
}
}
```
The "mqtt_only" can only be used for MQTT connections, which the user
"all" accepts standard, websocket and MQTT clients.
Here is what a configuration to enable MQTT looks like:
```
mqtt {
# Specify a host and port to listen for websocket connections
#
# listen: "host:port"
# It can also be configured with individual parameters,
# namely host and port.
#
# host: "hostname"
port: 1883
# TLS configuration section
#
# tls {
# cert_file: "/path/to/cert.pem"
# key_file: "/path/to/key.pem"
# ca_file: "/path/to/ca.pem"
#
# # Time allowed for the TLS handshake to complete
# timeout: 2.0
#
# # Takes the user name from the certificate
# #
# # verify_an_map: true
#}
# Authentication override. Here are possible options.
#
# authorization {
# # Simple username/password
# #
# user: "some_user_name"
# password: "some_password"
#
# # Token. The server will check the MQTT's password in the connect
# # protocol against this token.
# #
# # token: "some_token"
#
# # Time allowed for the client to send the MQTT connect protocol
# # after the TCP connection is established.
# #
# timeout: 2.0
#}
# If an MQTT client connects and does not provide a username/password and
# this option is set, the server will use this client (and therefore account).
#
# no_auth_user: "some_user_name"
# This is the time after which the server will redeliver a QoS 1 message
# sent to a subscription that has not acknowledged (PUBACK) the message.
# The default is 30 seconds.
#
# ack_wait: "1m"
# This limits the number of QoS1 messages sent to a session without receiving
# acknowledgement (PUBACK) from that session. MQTT specification defines
# a packet identifier as an unsigned int 16, which means that the maximum
# value is 65535. The default value is 1024.
#
# max_ack_pending: 100
}
```
Signed-off-by: Ivan Kozlovic <ivan@synadia.com>
