Inter-Cluster Communications

[goharahmed]

It will be awesome if we can remain within the boundaries of OpenSIPS & SIP protocol and send/retrieve data among the Clustered proxies.
The clusterer module and Bin proto are the first step towards that.

Ideally speaking, I want to send a SIP packet method("MYNAME") to the whole cluster or to a specific ID on the cluster. Similarly the receiving Proxy will be configured to recognize the method (MYNAME) and possibly reply with w/e data in headers/body it needs.

I see the developer functions in the Clusterer module doc but I want those functions in opensips.cfg

My Dream code(sample):

Sender Proxy:

$avp(body) = "Offender IP: " + $si;
$avp(my_method) = "BLOCK_IP";
send_all("$avp(my_method)","text/plain","$avp(body)");

$avp(body) = "Check IP: " + $si;
$avp(my_method) = "QUERY_IP";

send_to("$avp(my_method)","$avp(central_proxy_id)","text/plain","$avp(body)","$avp(reply)","resume_route");
....
route[resume_route] {
    if($avp(reply) = "BAD" ) {
        # Block this Caller: $si ; this is blocked Cluster-wide.
     }else if ($avp(reply) = "OK" ) {
           route(allowed_callers);
     }
}

Receiver Proxies:

if(check_address("1","$si")) 
   xlog("A Cluster Proxy Sent Something\n");
   if(is_method("BLOCK_IP")) { 
         $avp(offender) = $mb;
         # blocked IP $avp(offender) cluster-wide.
   }
   if(is_method("REGSYNC")) {
       $avp(subscriber) = $hdr(FROM);
       $avp(hosted_at) = $si;
      xlog("Subscriber $avp(subscriber) is now located at $avp(hosted_at)\n");
   }
   if(is_method("QUERY_IP")) {
         $avp(query_this) = $(mb{s.select,1,:});
         # Check IP $avp(query_this) From Local DB/Redis/Whatever
         if(found_blocked) {
               t_reply("200","BAD");
          }else {
               t_reply("200","OK");
         }
   }
}

Practically, all of the above is found in Kamailio as DMQ module for cluster communications, or additionally UAC module to compose a SIP packet...I'd love if I can do this within OpenSIPS w/o using rest_client, rabbitMQ and other external dependencies.

[rvlad-patrascu]

What you are trying to achieve, from what I understand (send/receive SIP messages between proxies with a custom method and body), is already possible through this tm function. Do you have any specific requirements that would benefit from using the clusterer module? Also, note that when using the clusterer module, the nodes communicate through a custom binary protocol (not SIP messages).

[goharahmed]

Hi Robert,

Thanks for your reply and it seems that I indeed missed to look at TM module for those custom method/reply functions. Yes they are helpful in getting one step closer, however, I can imagine a messy config file for more than 10 opensips in a global deployment.

The clusterer module would help me greatly in a sense if I have a shared clusterer table and now each proxy can broadcast things out to cluster w/o needing to manually code IP addresses(or some logic) anywhere. We can just name this module as sip_clusterer !

Its not necessary to use Binary protocol, all I'm trying to get at is that one way or another I get the capability to easily share/inquire some custom data right from opensips.cfg between a large group of opensips servers.

[rvlad-patrascu]

Yes, it seems like being able to share/inquire custom data in an OpenSIPS cluster directly from the script would be a nice feature indeed. The best approach however is to use existing mechanisms. This means sending the generic messages through the BIN protocol at clusterer level and triggering an event through the Event Interface in order to receive the messages.
Broadcasting would be straightforward but correlating messages with their "replies" would require some additional logic in the script though as it would be difficult to make this completely transparent by the script send function.
I will come up with an implementation for this.

[rvlad-patrascu]

Implemented in 1812a05. Check the documentation to see if this would work for you.

[goharahmed]

This is awesome, I will setup my VMware-cluster with this commit and take it for a test drive shortly. Best Regards,

…

-- *GOHAR AHMED * *Sr. VoIP Systems Software Engineer * www.saevolgo.ca <http://www.linkedin.com/pub/gohar-ahmed/35/81b/107> <https://www.facebook.com/saevolgo/> <https://twitter.com/saevolgo> <http://saevolgo.blogspot.com/> <https://www.packtpub.com/virtualization-and-cloud/implementing-citrix-xenserver-quickstarter> *- SaevolGo delivers reliable VoIP services - *

On Fri, Jul 28, 2017 at 1:24 PM, Robert Vladut Patrascu < ***@***.***> wrote: Implemented in 1812a05 <1812a05>. Check the documentation to see if this would work for you. — You are receiving this because you authored the thread. Reply to this email directly, view it on GitHub <#1119 (comment)>, or mute the thread <https://github.com/notifications/unsubscribe-auth/AJltfnP81kUzflZVAi3gjZpjsV50p2C2ks5sShk0gaJpZM4NUQVV> .

[goharahmed]

Hi Again,
I have started to work with the newly created functions and I'm absolutely loving it. Here is a sample config file which can be copied to multiple opensips boxes just changing required IP addresses in listen and clusterer module places.
https://github.com/goharahmed/opensips-clusterer/blob/f621259b8138c226601d607b297c2d57277ad8f9/opensips.cfg#L425

However, I need some clarification for the line pointed in the link above.

When I use the function
async(wait_for_event("E_CLUSTERER_RPL_RECEIVED", "$avp(filter"), "5"), rpl_resume);

As soon as the clustered boxes reply, say the very first one, will the above line Land me in event-route
E_CLUSTERER_RPL_RECEIVED or normal route referenced for async() i.e rpl_resume !?

In the sample code here: http://www.opensips.org/html/docs/modules/2.4.x/clusterer.html#idp5569696
The rpl_resume route is written but it doesnt exaplin from where it got the $avp(msg) ?

[goharahmed]

Another Observation; the event route gets me weird variables instead of numeric Cluster_id and SRC_ID.
Remember db_mode = 0 for clusterer module..

event_route[E_CLUSTERER_REQ_RECEIVED]{
        fetch_event_params("$avp(cluster_id);$avp(src_id);$avp(rcv_msg);$avp(rcv_tag)");
        $avp(state) = $(avp(rcv_msg){s.select,0,:});
        $avp(user) = $(avp(rcv_msg){s.select,1,:});
        $avp(proxy) = $(avp(rcv_msg){s.select,2,:});
        xlog("L_INFO","[CLUSTER-TALK] {MSG:$avp(rcv_msg)} Sender Proxy:$avp(proxy) SRC_ID:$avp(src_id) 
         User: $avp(user) State: $avp(state) ClusterId: $avp(cluster_id) TAG: $avp(rcv_tag) \n");

The XLOG line prints this:

[CLUSTER-TALK] {MSG:WHEREIS:8227:192.168.1.246} Sender Proxy:192.168.1.246 SRC_ID:1431784808 User: 8227 State: WHEREIS ClusterId: 32680 TAG: 1-1-nnXfD2GmAMXOQKeyUC6SWDJA

I can imagine that src_id and cluster_id are INT values; either I'm missing out something OR there is a bug in fetch_event_params.

here are the debug log lines:

DBG:proto_bin:bin_read_req: tcp_read_req end
DBG:proto_bin:bin_read_req: Using the global ( per process ) buff
DBG:proto_bin:bin_parse_headers: received a COMPLETE message
DBG:proto_bin:bin_handle_req: completely received a message
DBG:proto_bin:bin_handle_req: Nothing more to read on TCP conn 0x7fa8555746e0, currently in state 0
DBG:core:call_callbacks: binary Packet CMD: 6. Module: clusterer
DBG:clusterer:bin_rcv_cl_packets: received clusterer message from: 192.168.1.246:60851
DBG:clusterer:handle_cl_gen_msg: Received generic clusterer message
DBG:core:bin_pop_str: Popped: '1-1-nnXfD2GmAMXOQKeyUC6SWDJA' [28]
DBG:core:bin_pop_str: Popped: 'WHEREIS:8227:192.168.1.246' [26]
DBG:core:evi_param_set: adding int param
DBG:core:evi_param_set: adding int param
DBG:core:evi_param_set: adding string param
DBG:core:evi_param_set: adding string param
DBG:core:evi_raise_event_msg: found subscriber E_CLUSTERER_REQ_RECEIVED
DBG:event_route:scriptroute_fetch: Fetching parameters for event E_CLUSTERER_REQ_RECEIVED
DBG:event_route:scriptroute_fetch: Successfully fetched 4 parameters
[CLUSTER-TALK] {MSG:WHEREIS:8227:192.168.1.246} Sender Proxy:192.168.1.246 SRC_ID:1431784808 User: 8227 State: WHEREIS ClusterId: 32680 TAG: 1-1-nnXfD2GmAMXOQKeyUC6SWDJA

[rvlad-patrascu]

Hi,

That was indeed a bug introduced by some later code refactoring, thanks for pointing it out, I've pushed a fix.

And to clarify what you have previously asked, although I think that processing the cluster replies in the resume route of the async call (i.e "rpl_resume") should be enough, if you also use an event route for E_CLUSTERER_RPL_RECEIVED, both of the routes will be run, as they both subscribe to that event.

As for the $avp(msg) in "rpl_resume" route, because of the wait_for_event function, all of the event parameters of E_CLUSTERER_RPL_RECEIVED are exposed in the resume route via AVP variables as $avp(param_name).

Also, in the script you have provided, I noticed that you did: $var(tag) = $ci; before calling cluster_broadcast_req and then used $ci for the filtering AVP. This is not quite how it should be used. The send function will output an internally generated tag to a given variable, say "$var(tag)". The same variable should then be used to filter the E_CLUSTERER_RPL_RECEIVED events: $avp(filter) = "tag=" + $var(tag); because cluster_send_rpl should send back the same tag and this way it can be correlated with the request.

[goharahmed]

Thanks Robert,
I will do a git pull and retest the code. Hopefully I'll be able to send "replies" successfully this time. Once Im on a reply event then I will get to know about rpl_resume and event_route.

On the note about using "tag" - that was something new for me. Just asking; while I'm on a test platform I make just 1 call and get 1 reply and my "rpl_resume" route kicks in and gets all the variables as reply... BUT when Im in production and hundreds of calls will be sending broadcast and waiting for async reply...does that mean that a unique call with tag=111 may get resume by some other reply ,say tag=222 ??
OR
tag is just an additional check for script writer to be sure that the sent query gets the answer properly back ? it will always resume for the intended call and further we can validate like this:

if(    $var(tag) == $(avp(filter){s.select,1,=})   )
 
        I got my variables 

}else{

     something went wrong - I should never ever get resumed here for some other call !!

}

Additional Question:
I'm hoping that all my previous script AVPs and variables are still available in rpl_resume route as well ! _ let me know if I'm correct. ?

[rvlad-patrascu]

Well, that's exactly what the tag and filter are for, you don't have to do any additional checks in the script to correlate the replies you get in the resume route with the requests sent before the wait_for_event call. That function will take care to be resumed for the correct events. By setting $avp(filter) with the generated tag for that request you will land in the resume route when getting the reply exactly for that request. For example a call which triggered a cluster send with tag=111 will be resumed only for the reply with tag 111, a call with tag=222 for the corresponding reply and so on. You could also take a look at the event_routing module and wait_for_event function documentation to better understand the mechanisms.

In the resume route you will have acces to: all $avp variables, all changes in current SIP message and pseudo-varaibles like $ru, $rd, $ci etc. but no $var variables.

[goharahmed]

Hi Again,
Finally its working Here is a sample config that can deliver call from one proxy to any other in cluster.
https://github.com/goharahmed/opensips-clusterer/blob/master/opensips.cfg

Quick Note: I had to use permission module to add IP addresses of known Proxies ( to perform IP-AUTH) while the same data is pretty much in the clusterer module. A function from the clusterer module like
2.1.4. check_addr(cluster_id, su) will reduce that dependency as well.

Next step for me is to test its stability and stress test this one.

Thanks for your hard work for the OpenSIPS community.

[rvlad-patrascu]

Check out: 3f39e0e for such a function.