AMQP Transporter #72
AMQP Transporter #72
Conversation
🏁 Benchmark results📝 Moleculer common benchmarks⚡ Suite: Local call
⚡ Suite: Call with middlewares
⚡ Suite: Call with statistics & metrics
⚡ Suite: Remote call with FakeTransporter
Master detailed results{
"name": "Moleculer common benchmarks",
"suites": [
{
"name": "Local call",
"tests": [
{
"name": "broker.call (normal)",
"reference": true,
"stat": {
"duration": 5.000530275,
"cycle": 0,
"count": 2657000,
"avg": 0.0000018820211799021454,
"rps": 531343.6483493743,
"percent": 0
}
},
{
"name": "broker.call (with params)",
"fastest": true,
"stat": {
"duration": 5.000466823,
"cycle": 0,
"count": 2709000,
"avg": 0.0000018458718431155408,
"rps": 541749.4197821218,
"percent": 1.9583882229651408
}
}
]
},
{
"name": "Call with middlewares",
"tests": [
{
"name": "No middlewares",
"fastest": true,
"reference": true,
"stat": {
"duration": 5.000517228,
"cycle": 0,
"count": 2790000,
"avg": 0.00000179230008172043,
"rps": 557942.2833257361,
"percent": 0
}
},
{
"name": "5 middlewares",
"stat": {
"duration": 4.999572709,
"cycle": 0,
"count": 2604000,
"avg": 0.0000019199587976190474,
"rps": 520844.5104343416,
"percent": -6.649034138489213
}
}
]
},
{
"name": "Call with statistics & metrics",
"tests": [
{
"name": "No statistics",
"fastest": true,
"reference": true,
"stat": {
"duration": 5.000355966,
"cycle": 0,
"count": 2565000,
"avg": 0.0000019494565169590644,
"rps": 512963.4804883409,
"percent": 0
}
},
{
"name": "With metrics",
"stat": {
"duration": 5.013681836,
"cycle": 0,
"count": 432000,
"avg": 0.00001160574499074074,
"rps": 86164.22304624278,
"percent": -83.20265938537875
}
},
{
"name": "With statistics",
"stat": {
"duration": 5.001726296,
"cycle": 0,
"count": 574000,
"avg": 0.000008713808878048781,
"rps": 114760.37792372635,
"percent": -77.62796333679066
}
},
{
"name": "With metrics & statistics",
"stat": {
"duration": 5.089317055,
"cycle": 0,
"count": 329000,
"avg": 0.0000154690487993921,
"rps": 64645.2159385067,
"percent": -87.39769624984521
}
}
]
},
{
"name": "Remote call with FakeTransporter",
"tests": [
{
"name": "Remote call echo.reply",
"fastest": true,
"stat": {
"duration": 5.01190814,
"cycle": 0,
"count": 69000,
"avg": 0.00007263634985507247,
"rps": 13767.211623315985,
"percent": 0
}
}
]
}
],
"timestamp": 1502592737122,
"generated": "Sun Aug 13 2017 02:52:17 GMT+0000 (UTC)",
"elapsedMs": 46936
}
PR detailed results{
"name": "Moleculer common benchmarks",
"suites": [
{
"name": "Local call",
"tests": [
{
"name": "broker.call (normal)",
"fastest": true,
"reference": true,
"stat": {
"duration": 5.00058052,
"cycle": 0,
"count": 3300000,
"avg": 0.0000015153274303030302,
"rps": 659923.3802558588,
"percent": 0
}
},
{
"name": "broker.call (with params)",
"stat": {
"duration": 4.999847396,
"cycle": 0,
"count": 3124000,
"avg": 0.0000016004633149807937,
"rps": 624819.069977871,
"percent": -5.319452428610333
}
}
]
},
{
"name": "Call with middlewares",
"tests": [
{
"name": "No middlewares",
"reference": true,
"stat": {
"duration": 5.00078164,
"cycle": 0,
"count": 3276000,
"avg": 0.000001526490122100122,
"rps": 655097.5899039656,
"percent": 0
}
},
{
"name": "5 middlewares",
"fastest": true,
"stat": {
"duration": 5.000413262,
"cycle": 0,
"count": 3330000,
"avg": 0.0000015016256042042043,
"rps": 665944.9580509491,
"percent": 1.655840032715389
}
}
]
},
{
"name": "Call with statistics & metrics",
"tests": [
{
"name": "No statistics",
"fastest": true,
"reference": true,
"stat": {
"duration": 4.999767815,
"cycle": 0,
"count": 3345000,
"avg": 0.000001494698898355755,
"rps": 669031.0677956952,
"percent": 0
}
},
{
"name": "With metrics",
"stat": {
"duration": 5.004111904,
"cycle": 0,
"count": 422000,
"avg": 0.00001185808508056872,
"rps": 84330.64809415581,
"percent": -87.39510731960385
}
},
{
"name": "With statistics",
"stat": {
"duration": 5.003340497,
"cycle": 0,
"count": 776000,
"avg": 0.000006447603733247422,
"rps": 155096.38020144525,
"percent": -76.81776113739345
}
},
{
"name": "With metrics & statistics",
"stat": {
"duration": 5.146806354,
"cycle": 0,
"count": 311000,
"avg": 0.000016549216572347265,
"rps": 60425.82110327441,
"percent": -90.96815917646938
}
}
]
},
{
"name": "Remote call with FakeTransporter",
"tests": [
{
"name": "Remote call echo.reply",
"fastest": true,
"stat": {
"duration": 5.116487725,
"cycle": 0,
"count": 60000,
"avg": 0.00008527479541666667,
"rps": 11726.79447794434,
"percent": 0
}
}
]
}
],
"timestamp": 1502592847877,
"generated": "Sun Aug 13 2017 02:54:07 GMT+0000 (UTC)",
"elapsedMs": 47090
}
|
| opts = { amqp: { url: opts } }; | ||
|
|
||
| // Number of requests a broker will handle concurrently | ||
| if (typeof opts.amqp.prefetch !== "number") |
There was a problem hiding this comment.
This will break if opts is a string.
There was a problem hiding this comment.
@WoLfulus If opts was a string it should have already been reassigned to an object by this point. The first if-statement in the constructor should handle this case, I think.
There was a problem hiding this comment.
I'm sorry, I didn't see you reassigning opts at line 41, so yeah, it wont fail.
My bad :(
| opts.amqp.prefetch = 1; | ||
|
|
||
| // Number of milliseconds before an event expires | ||
| if (typeof opts.amqp.eventTimeToLive !== "number") |
|
Thanks @Nathan-Schwartz . Great job! I'm on holiday now, but I'm going to review it next week. |
| @@ -498,6 +501,13 @@ class Transit { | |||
| * @memberOf Transit | |||
| */ | |||
| publish(packet) { | |||
| if (this.subscribing) { | |||
There was a problem hiding this comment.
Is it solve an issue that publish a packet before subscriptions are ready
There was a problem hiding this comment.
Yes, thats correct.
src/transporters/amqp.js
Outdated
| this.logger.error("AMQP connection closed!", crashWorthy && err || ""); | ||
| }) | ||
| .on("blocked", (reason) => { | ||
| this.logger.warn("AMQP connection blocked!", reason); |
There was a problem hiding this comment.
From the docs I read that a RabbitMQ server (after version 3.2.0) can decide to block a connection.
Typically it will do this if there is some resource shortage, e.g., memory, and messages are published on the connection. - amqp.node
I don't know how common this error is, but I thought logging would help with debugging.
| }) | ||
| .filter(a => a); | ||
|
|
||
| xdescribe("Test AMQPTransporter events", () => { |
There was a problem hiding this comment.
xdescribe is the same as describe.skip, I did this for the integration tests I wrote so the describe block won't run. These tests require an AMQP server to be running at amqp://guest:guest@localhost:5672, so I don't think they would pass in the current CI environment.
There was a problem hiding this comment.
Thanks, I didn't know xdescribe :)
|
Integration test is perfect. But in case of dependency we can't running it in By the way, very nice PR. Congratulation 😉 👍 (I like that you followed my coding styles) |
|
I think another script makes sense 👍 Thank you! I really enjoyed working on Moleculer, it is an awesome project! |
|
@Nathan-Schwartz I'm trying to connect a remote RabbitMQ, but I got ECONNREFUSED from 127.0.0.1. // Create broker
let broker = new ServiceBroker({
namespace: "multi",
nodeID: process.argv[2] || "server-" + process.pid,
transporter: "amqp://guest:[email protected]:5672",
logger: console
});Err: |
Nathan-Schwartz
force-pushed
the
amqp-transporter
branch
from
730b537
to
fcae3b0
Compare
|
@icebob You are correct, I should have passed in |
1 similar comment
|
Thanks. Btw, how could you modify an old commit? :) |
|
I tested. It seems it's working properly ;) |
|
A minor difference. In all other transporter if the connection lost, trying reconnect in 5 sec. But in amqp, NATS: AMQP: And the process exited. |
|
@icebob I do it using It can create problems if you change a commit that is already merged though :) |
|
I wanted to implement reconnection, but its a bit tricky. Unfortunately right now
When the connection is lost, we could still try something like setTimeout(() => this.connect(), 5000);But the problem is that we don't know if all of the queues and exchanges were created successfully, and the exchanges might not be bound. To make sure that all exchanges, queues, and bindings are ready we would need Transit to call If we assume that all queues, bindings, and exchanges are fine, reconnecting could work in some cases. I think there would be some edge-cases we couldn't handle well, though. I tend to think that the transporter will behave more predictably if we don't try to reconnect. What do you think? |
|
The reconnecting logic is implemented in transit, not in transporters. Just need to reject the connect promise correctly. I'm trying to fix it. |
|
Great! I think it is ready 🙂 |
|
Merged. Very good job. Thanks again! Further task: It would be good to separate amqp integration test, skip it in jest when we call Brainstorming: What do you think. Because of RabbitMQ has load balancing solution. This new AMQP transporter can work without Moleculer built-in load balancing? I think transporter disable load balancing in broker. But I have no experience with AMQP and I can't consider it completely :( |
|
@Nathan-Schwartz What do you think? |
|
@icebob For the tests we could try a folder structure like this: and then we would ignore the thirdParty directory for the I think the AMQP transporter would work without Moleculer's load balancing. Are you thinking about making the transporter work with Moleculer load balancing, or disabling Moleculer's load balancing when the AMQP Transporter is used? |
|
Thanks @Nathan-Schwartz!
By the way within some days I will open a PR about AMQP optimization because it doesn't need to call |
|
@icebob Sounds good! I didn't know about Currently it is re-asserting the REQ queues for every INFO packet, but if a service is loaded dynamically none of the new actions will a REQ queue yet. So I think One way to avoid re-asserting existing queues would be to track them in |
|
The other problem if I destroy a service. How can I remove asserted REQ queues? Could you show an example? |
|
@icebob I don't think the REQ queues or any exchanges should be destroyed. They could be in use by other services and destroying them would break the other nodes. Note:
If there was only one node for a service, a REQ queue could be destroyed without breaking things. Even if it is possible to delete a queue without crashing nodes, I don't think we should do it. AMQP queues are often used as a Work Queue, so multiple nodes could be pulling from the same queue and there may be many messages waiting to be processed. If we delete the queue, we would delete all of the messages and this could be very problematic for a user. |
|
Thanks Nathan, It's good to know! |
|
I tried the Thanks in advance! |
|
@icebob It looks good 👍 My only comment would be that I don't think we should use It should be fine for the other packets since the messages will be removed after 5 seconds anyways. |
|
Ohh, I thought it would be deleted when it's empty and no consumers. :( |
|
@Nathan-Schwartz next time, please consider to use |
|
Thanks. I changed it in other branch |
|
@icebob I just found out about the |
|
@Nathan-Schwartz It would be good. Could you create a PR? |
|
Hi @Nathan-Schwartz, I rewrote the AMQP integration tests. I used multiple broker & transporters in single files instead of multi processes. Here is my tests: https://github.com/ice-services/moleculer/tree/next/test/integration/amqp What do you think, is it cover all test cases from your original tests? |
|
@icebob Good idea! It looks good, but there are a few small things I'm noticing. Would you be willing to open a PR so it is easier to review / comment on? |
|
@Nathan-Schwartz Thanks. I already merged it, but I think you can add comments to the PR. |
|
@icebob Sorry it took me so long to do a proper review, I'll try to be faster in the future. I started looking into the new AMQP Transporter and I think using the AMQP load-balancer and grouped events are cool ideas. I wanted to ping you because I'm noticing some issues. Two of the new tests for disabled load balancer fail:
The new integration tests don't include the following tests, and the old tests for these cases fail:
I tried to figure out why the tests fail and I found that we are no longer waiting for the action handler to complete before acknowledging messages. This means that I am happy to help resolve the above issues and write more tests in the new style, if you are open to it. Also, I want to suggest that we add a RabbitMQ server to Travis so that we can run integration tests before merging changes. Do you think that would be possible? Thanks! |
|
Hi @Nathan-Schwartz, thank you for your response. I would be happy if you can make the mentioned tests cases (would be fail). And after it I can start to fix the messageHandler code that the tests will be success. Could you help me to write the test cases? |
|
@icebob I see, thanks for the explanation. Sounds like a plan! I will start working on the test cases today 👍 What do you think about adding a RabbitMQ server to TravisCI? |
|
Great! RabbitMQ on Travis would be good, but I want to skip amqp integration tests from |
|
@icebob Sounds good to me! |
|
@icebob Sometimes I am getting an error in the test case I looked into it for a while and it seems to happen when publishing the I tried turning off It doesn't always fail but it could cause some inconvenience in CI, so I would be okay marking it as pending for now until we figure it out. |
Resolves #1
This Transporter is a bit more involved than some of the others because it needs to treat some packet types specially. Broadcasted packets need to use AMQP exchanges, but targeted packets only need a queue. This implementation also deals with Requests a bit differently than other Transporters do. Each action has its own work queue that multiple nodes can pull messages from. This means that messages are less likely to be lost (due to crashes / etc). Each node will consume more messages when it has an availability. The number of messages that can be handled concurrently by a node can be specified with the
amqp.prefetchoption.Requests and Responses don’t have a set time-to-live, but all other packet types do. Currently events’ time-to-live can be configured using the
amqp.eventTimeToLiveoption, but the other types are hardcoded at 5 seconds.Transporter Options:
AMQP Transport behaviour by topic
I’ve included unit tests and some integration tests. The integration tests are skipped right now because they require an amqp server to be running at
amqp://guest:guest@localhost:5672. Let me know if I should remove them. 👍Other notes:
istanbul ignorethe same way as in other files, but let me know if I should change anything.