broadcasting through omega-supreme only works on the same process

[siddo420]

this could possibly be a documentation bug or perhaps am doing something wrong.

through my tests, it was revealed that the broadcast messages are only sent to clients connected to the same node process and not to the "cluster" as is mentioned in the documentation.

Here is whats written: "Broadcast a message to all sparks in the cluster."

I tested by using two node processes behind HAProxy. I fired up two browser sessions and made sure clients were connected to separate node processes. I sent messages and only received them back on the sender but the other client that was connected to the other node process did not receive anything.

I killed one of the node processes and forced the client connected to it to use the other node process (behind HAProxy). This time I could see that messages were received on both clients (because both were connected to the same process now).

The code I used to broadcast message is:

                primus.forward.broadcast( data.message, function (err, result) {
                  logger.debug("In send message: " + JSON.stringify( result ) + 
                                         " error: " + JSON.stringify( err ));
                });

The debug line returned (after a few minutes of sending message).

debug: In send message: [null] error: {"code":"ECONNRESET","url":"http://192.168.1.9:3000/primus/omega/supreme","status":500,"body":{},"type":"broadcast","packet":{"sparks":"","msg":"hello"}}

Its the same error object that is returned no matter if the messages were delivered or not.

[lpinca]

@siddo420 I wrote a very simple test case some time ago: https://gist.github.com/lpinca/dac8f2efca54518555a4

Can you try to run that and see if it works for you? It does not use primus.forward.broadcast directly but you can tweak it to use that.

[siddo420]

I took the essence of your test case (rest is you creating clients that I do manually) and used it but same error. I wonder if I need to have a PUT route configured in express for omega-supreme (dont see that in omega's docs though).

primus.forward(servers, data, function (err, data) {
    if (err) throw err;
    console.log(data);
});

Perhaps, omega examples are incorrect ... trying more now.

[lpinca]

In that example I didn't use the convenience method that is added by metroplex, but you can do this:

primus.forward.broadcast(data, function (err, result) {
  if (err) throw err;
  console.log(result);
});

You don't need a route in express because primus requests are handled and answered by primus.

[siddo420]

thats exactly what I used before (see my first comment above).

Any chance you could add a new function to return all spark IDs linked to a given server address? It shouldn't be too hard for you I guess.

Redis 'smembers' call on

<namespace>:<http://....>:sparks"

[siddo420]

looks like something is wrong with omega-supreme because I used the following code this time (smembers included) to send messages to sparks.

primus.metroplex.servers( function( err, servers ){
   logger.info( 'registered servers:', servers );
   for( var i=0; i < servers.length; i++ ){
     redis.smembers( primusOptions.namespace + ":" + servers[i] + ":sparks", function( err, sparks ){
       logger.debug( "Returning Redis members: " + JSON.stringify(sparks) + JSON.stringify(err) );
       primus.forward( servers[i], { msg: data.message }, sparks, function (err, result) {
         logger.debug( "In message send: " + JSON.stringify(err) + JSON.stringify(result) );
       });
     });
   }
});

This should forward message to all sparks in all other hosts (except local).. however it instead returns:

registered servers: 0=http://192.168.1.9:3001
Returning Redis members: ["3VTVz6T3OpbML9GdAAAA"]null
In message send: {}{"ok":false,"send":0,"local":false}

Any ideas what might be wrong with omega-primus here?

[lpinca]

@siddo420 try to create a super simple test case to reproduce the issue, like the one I posted above, otherwise is too hard to debug.

Any chance you could add a new function to return all spark IDs linked to a given server address? It shouldn't be too hard for you I guess.

Yes this can be done, and as you said it's not hard, mind to create a pull request for this?

[siddo420]

it is simple i think, scenario is a bit complicated so test case has to be a bit complicated as well.

Dont know how to create a pull request. However, here is the function you need to get sparks from a server.

redis.smembers( namespace + ":" + server + ":sparks", function( err, sparks ){})

[lpinca]

The code from your previous comment is wrong:

primus.metroplex.servers( function( err, servers ){
   logger.info( 'registered servers:', servers );
   for( var i=0; i < servers.length; i++ ){
     redis.smembers( primusOptions.namespace + ":" + servers[i] + ":sparks", function( err, sparks ){
       logger.debug( "Returning Redis members: " + JSON.stringify(sparks) + JSON.stringify(err) );
       primus.forward( servers[i], { msg: data.message }, sparks, function (err, result) {
         logger.debug( "In message send: " + JSON.stringify(err) + JSON.stringify(result) );
       });
     });
   }
});

servers[i] is undefined because you are using it when i === servers.length. To avoid this problem, rewrite it like this:

primus.metroplex.servers(function (err, servers){
   logger.info('registered servers:', servers);
   servers.forEach(function (server) {
    redis.smembers(primusOptions.namespace + ":" + server + ":sparks", function (err, sparks) {
       logger.debug("Returning Redis members: " + JSON.stringify(sparks) + JSON.stringify(err));
       primus.forward(server, { msg: data.message }, sparks, function (err, result) {
         logger.debug("In message send: " + JSON.stringify(err) + JSON.stringify(result));
       });
     });
   });
});

[siddo420]

Yes, I had figured that out already and corrected it but that did not resolve the issue either.

I also wrote a very simple test case but still I only see messages on the clients connected to same node process only (and not the other host+process in cluster).

"mp" is the namespace I used.

redis.hkeys( "mp:sparks", function( err, sparks ){
  primus.forward( "http://192.168.1.9:3000", { msg: data.message }, sparks, function (err, result) {
  });
  primus.forward( "http://192.168.1.9:3001", { msg: data.message }, sparks, function (err, result) {
  });
});

redis returns all sparks on all hosts and I manually used the two server addresses in primus.forward call. Keep getting the following messages:

{"code":"ECONNRESET","url":"http://192.168.1.9:3001/primus/omega/supreme","status":500,"body":{},"type":"sparks","packet":{"sparks":["RRVU2OUn-A-y7f5FAAAA"],"msg":{"msg":"no problem"}}}

[lpinca]

The problem is that for some reasons, unknown to me, the request fails. Try to send the message like this:

curl -H "Content-Type: application/json" -X PUT -d '{"msg":{"msg":"no problem"}}' http://omega:supreme@192.168.1.9:3000/primus/omega/supreme

[siddo420]

it turns out that my custom authorization module was blocking the PUT requests, its been fixed now.

Thanks for working with me on this issue.

[3rd-Eden]

@lpinca in regards to the feature request, it seems sane and useful to me.

[lpinca]

Ok, given that thre is already a method called sparks that returns the servers of the given spark ids, how would you call this new method to return the spark ids for the given server?

[3rd-Eden]

That is an excellent question. I have no idea. We also already have a servers to get a list of servers. So most sane API names are already taken here. Maybe just a simple find method? If you have any better brainfarts please share :p

On Thursday, July 9, 2015 at 7:59 AM, Luigi Pinca wrote:

Ok, given that thre is already a method called sparks that returns the servers of the given spark ids, how would you call this new method to return the spark ids for the given server?

—
Reply to this email directly or view it on GitHub (#6 (comment)).

[lpinca]

I was thinking about something like this to get the servers:

primus.metroplex.servers([self], [sparks], fn);

Where self is a boolean to include the current server in the results and sparks an array of spark ids or a string representing a single spark id. Both self and sparks are optional and if sparks is provided self is ignored defaulting to true.

Another option is to check the type of first argument of the sparks method. If it is an array assume that we want to get the servers of the given spark ids. If it is a string assume that we want to get the spark ids of the server identified by this string.

primus.metroplex.sparks(/*array of spark ids | server string*/, fn)

The first option is a big breaking change but I don't think that it is a problem, unless we revert the ioredis pull request, master already contains breaking changes.

The problem is that I like none of the above solutions 😕.