Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

js: update account limits #357

Merged
merged 1 commit into from
Sep 27, 2022
Merged

js: update account limits #357

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
31 changes: 29 additions & 2 deletions nats/js/api.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -461,6 +461,24 @@ class AccountLimits(Base):
max_storage: int
max_streams: int
max_consumers: int
max_ack_pending: int
memory_max_stream_bytes: int
storage_max_stream_bytes: int
max_bytes_required: bool


@dataclass
class Tier(Base):
memory: int
storage: int
streams: int
consumers: int
limits: AccountLimits

@classmethod
def from_response(cls, resp: Dict[str, Any]):
cls._convert(resp, 'limits', AccountLimits)
return super().from_response(resp)


@dataclass
Expand All @@ -478,20 +496,29 @@ class AccountInfo(Base):
References:
* `Account Information <https://docs.nats.io/jetstream/administration/account#account-information>`_
"""

# NOTE: These fields are shared with Tier type as well.
memory: int
storage: int
streams: int
consumers: int
limits: AccountLimits

api: APIStats
domain: Optional[str] = None
tiers: Optional[Dict[str, Tier]] = None

@classmethod
def from_response(cls, resp: Dict[str, Any]):
cls._convert(resp, 'limits', AccountLimits)
cls._convert(resp, 'api', APIStats)
return super().from_response(resp)
info = super().from_response(resp)
tiers = resp.get('tiers', None)
if tiers:
result = {}
for k, v in tiers.items():
result[k] = Tier.from_response(v)
info.tiers = result
return info


@dataclass
Expand Down
22 changes: 22 additions & 0 deletions tests/conf/js-limits.conf
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,22 @@

jetstream: {
domain: "test-domain"
}

no_auth_user: "foo"

accounts: {
A {
users: [{ user: "foo" }]
jetstream: {
max_mem: 64MB,
max_file: 32MB,
max_streams: 10,
max_consumers: 20,
max_ack_pending: 100,
memory_max_stream_bytes: 2048,
store_max_stream_bytes: 4096,
max_stream_bytes: true
}
}
}
180 changes: 180 additions & 0 deletions tests/test_js.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@
import time
import unittest
import uuid
import json

import pytest
import nats
Expand Down Expand Up @@ -2567,3 +2568,182 @@ async def test_number_of_consumer_replicas(self):
assert cons.config.num_replicas == 1

await nc.close()


class AccountLimitsTest(SingleJetStreamServerLimitsTestCase):

@async_test
async def test_account_limits(self):
nc = await nats.connect()

js = nc.jetstream()

with pytest.raises(BadRequestError) as err:
await js.add_stream(name="limits", subjects=["limits"])
assert err.value.err_code == 10113
assert err.value.description == "account requires a stream config to have max bytes set"

with pytest.raises(BadRequestError) as err:
await js.add_stream(
name="limits", subjects=["limits"], max_bytes=65536
)
assert err.value.err_code == 10122
assert err.value.description == "stream max bytes exceeds account limit max stream bytes"

si = await js.add_stream(
name="limits", subjects=["limits"], max_bytes=128
)
assert si.config.max_bytes == 128

await js.publish(
"limits",
b'AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA'
)
si = await js.stream_info("limits")
assert si.state.messages == 1

for i in range(0, 5):
await js.publish("limits", b'A')

expected = nats.js.api.AccountInfo(
memory=0,
storage=111,
streams=1,
consumers=0,
limits=nats.js.api.AccountLimits(
max_memory=67108864, # 64MB
max_storage=33554432, # 32MB
max_streams=10,
max_consumers=20,
max_ack_pending=100,
memory_max_stream_bytes=2048,
storage_max_stream_bytes=4096,
max_bytes_required=True
),
api=nats.js.api.APIStats(total=4, errors=2),
domain='test-domain',
tiers=None
)
info = await js.account_info()
assert expected == info

# Messages are limited.
js = nc.jetstream(domain="test-domain")
si = await js.stream_info("limits")
assert si.state.messages == 3

# Check unmarshalling response with Tiers:
blob = """{
"type": "io.nats.jetstream.api.v1.account_info_response",
"memory": 0,
"storage": 6829550,
"streams": 1,
"consumers": 0,
"limits": {
"max_memory": 0,
"max_storage": 0,
"max_streams": 0,
"max_consumers": 0,
"max_ack_pending": 0,
"memory_max_stream_bytes": 0,
"storage_max_stream_bytes": 0,
"max_bytes_required": false
},
"domain": "ngs",
"api": {
"total": 6,
"errors": 0
},
"tiers": {
"R1": {
"memory": 0,
"storage": 6829550,
"streams": 1,
"consumers": 0,
"limits": {
"max_memory": 0,
"max_storage": 2000000000000,
"max_streams": 100,
"max_consumers": 1000,
"max_ack_pending": -1,
"memory_max_stream_bytes": -1,
"storage_max_stream_bytes": -1,
"max_bytes_required": true
}
},
"R3": {
"memory": 0,
"storage": 0,
"streams": 0,
"consumers": 0,
"limits": {
"max_memory": 0,
"max_storage": 500000000000,
"max_streams": 25,
"max_consumers": 250,
"max_ack_pending": -1,
"memory_max_stream_bytes": -1,
"storage_max_stream_bytes": -1,
"max_bytes_required": true
}
}
}}
"""

expected = nats.js.api.AccountInfo(
memory=0,
storage=6829550,
streams=1,
consumers=0,
limits=nats.js.api.AccountLimits(
max_memory=0,
max_storage=0,
max_streams=0,
max_consumers=0,
max_ack_pending=0,
memory_max_stream_bytes=0,
storage_max_stream_bytes=0,
max_bytes_required=False
),
api=nats.js.api.APIStats(total=6, errors=0),
domain='ngs',
tiers={
'R1':
nats.js.api.Tier(
memory=0,
storage=6829550,
streams=1,
consumers=0,
limits=nats.js.api.AccountLimits(
max_memory=0,
max_storage=2000000000000,
max_streams=100,
max_consumers=1000,
max_ack_pending=-1,
memory_max_stream_bytes=-1,
storage_max_stream_bytes=-1,
max_bytes_required=True
)
),
'R3':
nats.js.api.Tier(
memory=0,
storage=0,
streams=0,
consumers=0,
limits=nats.js.api.AccountLimits(
max_memory=0,
max_storage=500000000000,
max_streams=25,
max_consumers=250,
max_ack_pending=-1,
memory_max_stream_bytes=-1,
storage_max_stream_bytes=-1,
max_bytes_required=True
)
)
}
)
info = nats.js.api.AccountInfo.from_response(json.loads(blob))
assert expected == info
await nc.close()
22 changes: 22 additions & 0 deletions tests/utils.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -489,6 +489,28 @@ def tearDown(self):
self.loop.close()


class SingleJetStreamServerLimitsTestCase(unittest.TestCase):

def setUp(self):
self.server_pool = []
self.loop = asyncio.new_event_loop()

server = NATSD(
port=4222,
with_jetstream=True,
config_file=get_config_file("conf/js-limits.conf")
)
self.server_pool.append(server)
for natsd in self.server_pool:
start_natsd(natsd)

def tearDown(self):
for natsd in self.server_pool:
natsd.stop()
shutil.rmtree(natsd.store_dir)
self.loop.close()


def start_natsd(natsd: NATSD):
natsd.start()

Expand Down