modules/cdp: new RPC commands to disable/enable a peer dynamically

	- this is useful to disable a particular diameter peer in a cluster

modules/cdp/cdp_rpc.c

@@ -0,0 +1,66 @@
+/*
+ * rpc.c
+ *
+ *  Created on: 27 May 2014
+ *      Author: jaybeepee
+ */
+#include "cdp_rpc.h"
+#include "peermanager.h"
+#include "peerstatemachine.h"
+#include "receiver.h"
+#include "../../str.h"
+#include "../../dprint.h"
+
+extern dp_config *config;
+
+static const char* cdp_rpc_disable_peer_doc[2] 	= 	{"disable diameter peer", 0 };
+static const char* cdp_rpc_enable_peer_doc[2] 	= 	{"enable diameter peer", 0 };
+
+
+static void cdp_rpc_enable_peer(rpc_t* rpc, void* ctx)
+{
+	peer *cdp_peer;
+	str peer_fqdn;
+
+	if (rpc->scan(ctx, "S", &peer_fqdn) < 1) {
+		rpc->fault(ctx, 400, "required peer fqdn argument");
+		return;
+	}
+
+	cdp_peer = get_peer_by_fqdn(&peer_fqdn);
+	if (cdp_peer != NULL) {
+		LM_DBG("Enabling CDP Peer: [%.*s]\n", peer_fqdn.len, peer_fqdn.s);
+		cdp_peer->disabled = 0;
+		return;
+	}
+	rpc->fault(ctx, 400, "peer not found");
+	return;
+}
+
+static void cdp_rpc_disable_peer(rpc_t* rpc, void* ctx)
+{
+	peer *cdp_peer;
+	str peer_fqdn;
+
+	if (rpc->scan(ctx, "S", &peer_fqdn) < 1) {
+		rpc->fault(ctx, 400, "required peer fqdn argument");
+		return;
+	}
+	cdp_peer = get_peer_by_fqdn(&peer_fqdn);
+	if (cdp_peer != NULL) {
+		LM_DBG("Disabling CDP peer: [%.*s]\n", peer_fqdn.len, peer_fqdn.s);
+		cdp_peer->disabled = 1;
+		return;
+	}
+
+	rpc->fault(ctx, 400, "peer not found");
+	return;
+
+}
+
+rpc_export_t cdp_rpc[] = {
+	{"cdp.disable_peer",	cdp_rpc_disable_peer,   cdp_rpc_disable_peer_doc,   0},
+	{"cdp.enable_peer",   	cdp_rpc_enable_peer,   	cdp_rpc_enable_peer_doc,   	0},
+	{0, 0, 0, 0}
+};
+

modules/cdp/cdp_rpc.h

@@ -0,0 +1,16 @@
+/*
+ * rpc.h
+ *
+ *  Created on: 27 May 2014
+ *      Author: jaybeepee
+ */
+
+#ifndef RPC_H_
+#define RPC_H_
+
+#include "../../rpc.h"
+#include "config.h"
+
+extern rpc_export_t cdp_rpc[];
+
+#endif /* RPC_H_ */

modules/cdp/config.h

@@ -124,7 +124,7 @@ typedef struct {
 	acceptor_config *acceptors;	/**< list of acceptors */
 	int acceptors_cnt;			/**< size of the list of acceptors */
 	
-	app_config *applications;	/**< list of supporter applications */
+	app_config *applications;	/**< list of supported applications */
 	int applications_cnt;		/**< size of list of supported applications*/
 
 	int *supported_vendors;		/**< list of supported vendor ids */

modules/cdp/diameter_comm.c

@@ -138,6 +138,7 @@ AAAReturnCode AAASendMessage(
 		LM_ERR("AAASendMessage(): Can't find a suitable connected peer in the routing table.\n");
 		goto error;
 	}
+	LM_DBG("Found diameter peer [%.*s] from routing table\n", p->fqdn.len, p->fqdn.s);
 	if (p->state!=I_Open && p->state!=R_Open){
 		LM_ERR("AAASendMessage(): Peer not connected to %.*s\n",p->fqdn.len,p->fqdn.s);
 		goto error;

modules/cdp/doc/cdp_admin.xml

@@ -15,9 +15,9 @@
     <para>CDP (C Diameter Peer) allows Diameter communication to and from
     Kamailio. Most of the code is inherited from DISC
     http://developer.berlios.de/projects/disc/ and OpenIMS and modified for
-    use within Kamailio. A few improvements and new functionality has been added
-    along the way, for example, threshold reporting on Diameter calls that are
-    serviced above a certain threshold.</para>
+    use within Kamailio. A few improvements and new functionality has been
+    added along the way, for example, threshold reporting on Diameter calls
+    that are serviced above a certain threshold.</para>
   </section>
 
   <section>
@@ -137,6 +137,24 @@ modparam("cdp", "latency_threshold", 1000)
     </section>
   </section>
 
+  <section>
+    <title>RPC Commands</title>
+
+    <para>exported RPC commands.</para>
+
+    <section>
+      <title>cdp.disable_peer</title>
+
+      <para>instantly disable a particular diameter peer.</para>
+    </section>
+
+    <section>
+      <title>cdp.enable_peer</title>
+
+      <para>enabe/re-enable a diameter peer</para>
+    </section>
+  </section>
+
   <section>
     <title>Configuration Examples</title>
 

modules/cdp/mod.c

@@ -51,7 +51,9 @@
 #include "diameter_peer.h"
 #include "config.h"
 #include "cdp_load.h"
-
+#include "cdp_rpc.h"
+#include "../../rpc.h"
+#include "../../rpc_lookup.h"
 #include "../../cfg/cfg_struct.h"
 
 MODULE_VERSION
@@ -198,6 +200,10 @@ struct module_exports exports = {
  */
 static int cdp_init( void )
 {
+	if (rpc_register_array(cdp_rpc) != 0) {
+		LM_ERR("failed to register RPC commands for CDP module\n");
+		return -1;
+	}
 #ifdef STATISTICS
 	/* register statistics */
 	if ( register_stat("cdp", "replies_response_time", &replies_response_time,0 )!=0 ) {

modules/cdp/peer.h

@@ -112,6 +112,7 @@ typedef struct _peer_t{
 	
 	time_t activity;		/**< timestamp of last activity */
 	int is_dynamic;			/**< whether this peer was accepted although it was not initially configured */
+	int disabled;			/**< administratively enable/disable a peer - ie remove/re-add from service dynamically */
 	int waitingDWA;			/**< if a Diameter Watch-dog Request was sent out and waiting for an answer */
 	
 	str send_pipe_name;		/**< pipe to signal messages to be sent out*/

modules/cdp/peermanager.c

@@ -264,6 +264,16 @@ int peer_timer(time_t now,void *ptr)
 	while(p){
 		lock_get(p->lock);
 		n = p->next;
+
+		if (p->disabled && (p->state != Closed || p->state != Closing)) {
+			LM_DBG("Peer [%.*s] has been disabled - shutting down\n", p->fqdn.len, p->fqdn.s);
+			if (p->state == I_Open) sm_process(p, Stop, 0, 1, p->I_sock);
+			if (p->state == R_Open) sm_process(p, Stop, 0, 1, p->R_sock);
+			lock_release(p->lock);
+			p = n;
+			continue;
+		}
+
 		if (p->activity+config->tc<=now){
 			LM_INFO("peer_timer(): Peer %.*s \tState %d \n",p->fqdn.len,p->fqdn.s,p->state);
 			switch (p->state){
@@ -274,8 +284,10 @@ int peer_timer(time_t now,void *ptr)
 						free_peer(p,1);
 						break;
 					}
-					touch_peer(p);
-					sm_process(p,Start,0,1,0);
+					if (!p->disabled) {
+						touch_peer(p);
+						sm_process(p,Start,0,1,0);
+					}
 					break;
 				/* timeouts */
 				case Wait_Conn_Ack:
@@ -293,10 +305,12 @@ int peer_timer(time_t now,void *ptr)
 						p->waitingDWA = 0;
 						if (p->state==I_Open) sm_process(p,I_Peer_Disc,0,1,p->I_sock);
 						if (p->state==R_Open) sm_process(p,R_Peer_Disc,0,1,p->R_sock);
+						LM_WARN("Inactivity on peer [%.*s] and no DWA, Closing peer...\n", p->fqdn.len, p->fqdn.s);
 					} else {
 						p->waitingDWA = 1;
 						Snd_DWR(p);
 						touch_peer(p);
+						LM_WARN("Inactivity on peer [%.*s], sending DWR... - if we don't get a reply, the peer will be closed\n", p->fqdn.len, p->fqdn.s);
 					}
 					break;
 				/* ignored states */

modules/cdp/routing.c

@@ -86,7 +86,7 @@ peer* get_first_connected_route(routing_entry *r,int app_id,int vendor_id)
 		else
 			LM_DBG("The peer %.*s state is %s\n",i->fqdn.len,i->fqdn.s,
 				(p->state==I_Open||p->state==R_Open)?"opened":"closed");
-		if (p && (p->state==I_Open || p->state==R_Open) && peer_handles_application(p,app_id,vendor_id)) {			
+		if (p && !p->disabled && (p->state==I_Open || p->state==R_Open) && peer_handles_application(p,app_id,vendor_id)) {
 			LM_DBG("The peer %.*s matches - will forward there\n",i->fqdn.len,i->fqdn.s);
 			return p;
 		}
@@ -112,6 +112,8 @@ peer* get_routing_peer(AAAMessage *m)
 	routing_realm *rr;
 	int app_id=0,vendor_id=0;
 	
+	LM_DBG("getting diameter routing peer for realm: [%.*s]\n", m->dest_realm->data.len, m->dest_realm->data.s);
+
 	app_id = m->applicationId;	
 	avp = AAAFindMatchingAVP(m,0,AVP_Vendor_Specific_Application_Id,0,AAA_FORWARD_SEARCH);
 	if (avp){
@@ -178,6 +180,7 @@ peer* get_routing_peer(AAAMessage *m)
 	}
 	/* if not found in the realms or no destination_realm, 
 	 * get the first connected host in default routes */
+	LM_DBG("no routing peer found, trying default route\n");
 	p = get_first_connected_route(config->r_table->routes,app_id,vendor_id);
 	if (!p){
 		LM_ERR("get_routing_peer(): No connected DefaultRoute peer found for app_id %d and vendor id %d.\n",