monitube2-data_conn.c

Go to the documentation of this file.
00001 /*
00002  * $Id: monitube2-data_conn.c 347265 2009-11-19 13:55:39Z kdickman $
00003  *
00004  * This code is provided as is by Juniper Networks SDK Developer Support.
00005  * It is provided with no warranties or guarantees, and Juniper Networks
00006  * will not provide support or maintenance of this code in any fashion.
00007  * The code is provided only to help a developer better understand how
00008  * the SDK can be used.
00009  *
00010  * Copyright (c) 2009, Juniper Networks, Inc.
00011  * All rights reserved.
00012  */
00013 
00021 #include "monitube2-data_main.h"
00022 #include <jnx/pconn.h>
00023 
00024 #include "monitube2-data_conn.h"
00025 #include "monitube2-data_config.h"
00026 
00027 
00028 /*** Constants ***/
00029 
00033 #define RETRY_CONNECT 60
00034 #define CONNECT_RETRIES    1  
00035 
00036 
00037 /*** Data Structures ***/
00038 
00039 
00044 typedef struct notification_msg_s {
00045     msg_type_e     action;   
00046     flow_stat_t *  message;  
00047     TAILQ_ENTRY(notification_msg_s) entries;    
00048 } notification_msg_t;
00049 
00050 
00054 static TAILQ_HEAD(notification_buffer_s, notification_msg_s) notification_msgs =
00055             TAILQ_HEAD_INITIALIZER(notification_msgs);
00056 
00057 static msp_spinlock_t msgs_lock;     
00058 
00059 static pconn_client_t * mgmt_client; 
00060 
00061 static evTimerID  mgmt_timer_id;     
00062 
00066 extern evContext * ctx;
00067 
00068 /*** STATIC/INTERNAL Functions ***/
00069 
00070 
00074 static void connect_mgmt(evContext ectx, void * uap,
00075     struct timespec due, struct timespec inter);
00076 
00077 
00094 static void
00095 process_notifications_messages(evContext ectx UNUSED, void * uap  UNUSED,
00096             struct timespec due UNUSED, struct timespec inter UNUSED)
00097 {
00098     notification_msg_t * msg;
00099     int rc = 0;
00100 
00101     if(mgmt_client == NULL) { // don't bother doing anything yet
00102         return;
00103     }
00104 
00105     INSIST_ERR(msp_spinlock_lock(&msgs_lock) == MSP_OK);
00106 
00107     while((msg = TAILQ_FIRST(&notification_msgs)) != NULL) {
00108 
00109         TAILQ_REMOVE(&notification_msgs, msg, entries);
00110 
00111         INSIST_ERR(msp_spinlock_unlock(&msgs_lock) == MSP_OK);
00112 
00113         rc = pconn_client_send(mgmt_client, msg->action, msg->message,
00114                 sizeof(flow_stat_t));
00115 
00116         if(rc != PCONN_OK) {
00117             // put message back in and process soon
00118 
00119             CLOG(LOG_ERR, "%s: Failed to send message to mgmt component."
00120                     " Error: %d", __func__, rc);
00121 
00122             INSIST_ERR(msp_spinlock_lock(&msgs_lock) == MSP_OK);
00123 
00124             TAILQ_INSERT_HEAD(&notification_msgs, msg, entries);
00125 
00126             INSIST_ERR(msp_spinlock_unlock(&msgs_lock) == MSP_OK);
00127 
00128             if(evSetTimer(*ctx, process_notifications_messages, NULL,
00129                     evAddTime(evNowTime(), evConsTime(5, 0)),
00130                     evConsTime(0, 0), NULL)) {
00131 
00132                 CLOG(LOG_EMERG, "%s: evSetTimer() failed! Will not be "
00133                         "able to process buffered notifications", __func__);
00134             }
00135         } else {
00136             free(msg->message);
00137             free(msg);
00138         }
00139 
00140         INSIST_ERR(msp_spinlock_lock(&msgs_lock) == MSP_OK);
00141     }
00142 
00143     INSIST_ERR(msp_spinlock_unlock(&msgs_lock) == MSP_OK);
00144 }
00145 
00146 
00151 static void
00152 process_notifications(void)
00153 {
00154     if(evSetTimer(*ctx, process_notifications_messages, NULL,
00155             evConsTime(0,0), evConsTime(0, 0), NULL)) {
00156 
00157         CLOG(LOG_EMERG, "%s: evSetTimer() failed! Will not be "
00158                 "able to process buffered notifications", __func__);
00159     }
00160 }
00161 
00174 static void
00175 mgmt_client_connection(pconn_client_t * client,
00176                       pconn_event_t event,
00177                       void * cookie UNUSED)
00178 {
00179     INSIST_ERR(client == mgmt_client);
00180 
00181     switch (event) {
00182 
00183     case PCONN_EVENT_ESTABLISHED:
00184 
00185         // clear the retry timer
00186         if(evTestID(mgmt_timer_id)) {
00187             evClearTimer(*ctx, mgmt_timer_id);
00188             evInitID(&mgmt_timer_id);
00189         }
00190 
00191         clear_config(); // we expect to get a fresh copy of everything
00192         
00193         CLOG(LOG_INFO, "%s: Connected to the mgmt component on the RE",
00194                         __func__);
00195 
00196         break;
00197 
00198     case PCONN_EVENT_FAILED:
00199 
00200         CLOG(LOG_ERR, "%s: Received a PCONN_EVENT_FAILED event", __func__);
00201         
00202         // NO break; falls thru
00203         
00204     case PCONN_EVENT_SHUTDOWN:
00205 
00206                 CLOG(LOG_INFO, "%s: Disconnected from the mgmt component on the RE",
00207                                 __func__);
00208 
00209         mgmt_client = NULL;
00210                 clear_config();
00211 
00212         // Reconnect to it if timer not going
00213         if(!evTestID(mgmt_timer_id) &&
00214            evSetTimer(*ctx, connect_mgmt, NULL, evConsTime(0,0),
00215                evConsTime(RETRY_CONNECT, 0), &mgmt_timer_id)) {
00216 
00217             CLOG(LOG_EMERG, "%s: Failed to initialize a re-connect timer to "
00218                 "reconnect to the mgmt component", __func__);
00219         }
00220 
00221         break;
00222 
00223     default:
00224         CLOG(LOG_ERR, "%s: Received an unknown pconn event", __func__);
00225     }
00226 }
00227 
00228 
00244 static status_t
00245 mgmt_client_message(pconn_client_t * session,
00246                     ipc_msg_t * msg,
00247                     void * cookie UNUSED)
00248 {
00249     struct in_addr ia;
00250 
00251     INSIST_ERR(session == mgmt_client);
00252     INSIST_ERR(msg != NULL);
00253 
00254     switch(msg->subtype) { // a msg_type_e
00255     
00256     /* subtypes:
00257 
00258     MSG_DELETE_ALL = 1,        ///< delete all configured policies, no msg data
00259     MSG_DELETE_SS,             ///< delete policies for a service set
00260     MSG_APPLY_RULE,            ///< apply rule to a service set
00261     MSG_REMOVE_RULE,           ///< remove rule from service set
00262     MSG_CONF_RULE_ACTION,      ///< create/update rule's action,
00263     MSG_DELETE_RULE,           ///< delete rule and remove from any ss
00264     MSG_CONF_RULE_MATCH_ADDR,  ///< create/update rule's match address
00265     MSG_DELETE_RULE_MATCH_ADDR ///< delete rule's match address
00266     
00267     */
00268 
00269     case MSG_DELETE_ALL:
00270 
00271         INSIST_ERR(msg->length == 0);
00272         
00273         CLOG(LOG_INFO, "%s: Received a configuration update: Clear "
00274                         "configuration", __func__);
00275         
00276         clear_config();
00277 
00278         break;
00279 
00280     case MSG_DELETE_SS:
00281     {
00282         del_ss_t * data = (del_ss_t *)msg->data;
00283 
00284         INSIST_ERR(msg->length == sizeof(del_ss_t));
00285 
00286         data->ss_id = ntohs(data->ss_id);
00287         data->gen_num = ntohl(data->gen_num);
00288         data->svc_id = ntohl(data->svc_id);
00289         
00290         CLOG(LOG_INFO, "%s: Received a configuration update: Delete service-set"
00291                 " %d", __func__, data->ss_id);
00292 
00293         delete_serviceset(data->ss_id, data->gen_num, data->svc_id);
00294 
00295         break;
00296     }
00297     case MSG_APPLY_RULE:
00298     {
00299         apply_rule_t * data = (apply_rule_t *)msg->data;
00300 
00301         INSIST_ERR(msg->length == sizeof(apply_rule_t) + 
00302                         ntohs(data->rule_name_len));
00303         
00304         data->ss_id = ntohs(data->ss_id);
00305         data->gen_num = ntohl(data->gen_num);
00306         data->svc_id = ntohl(data->svc_id);
00307 
00308         CLOG(LOG_INFO, "%s: Received a configuration update: Apply rule %s to "
00309                 "ssid: %d", __func__, data->rule_name, data->ss_id);
00310 
00311         apply_rule(data->rule_name, data->ss_id, data->gen_num, data->svc_id);
00312 
00313         break;
00314     }
00315     case MSG_REMOVE_RULE:
00316     {
00317         apply_rule_t * data = (apply_rule_t *)msg->data;
00318 
00319         INSIST_ERR(msg->length == sizeof(apply_rule_t) + 
00320                         ntohs(data->rule_name_len));
00321         
00322         data->ss_id = ntohs(data->ss_id);
00323         data->gen_num = ntohl(data->gen_num);
00324         data->svc_id = ntohl(data->svc_id);
00325 
00326         CLOG(LOG_INFO, "%s: Received a configuration update: Remove rule %s "
00327                 "from ssid: %d", __func__, data->rule_name, data->ss_id);
00328 
00329         remove_rule(data->rule_name, data->ss_id, data->gen_num, data->svc_id);
00330         break;
00331     }
00332     case MSG_CONF_RULE_ACTION:
00333     {
00334         update_rule_action_t * data = (update_rule_action_t *)msg->data;
00335 
00336         INSIST_ERR(msg->length == sizeof(update_rule_action_t) + 
00337                         ntohs(data->rule_name_len));
00338         
00339         data->rate = ntohl(data->rate);
00340         ia.s_addr = data->redirect;
00341 
00342         CLOG(LOG_INFO, "%s: Received a configuration update: Update rule %s "
00343                         "with rate %d and redirect %s", __func__, data->rule_name,
00344                         data->rate, inet_ntoa(ia));
00345 
00346         update_rule(data->rule_name, data->rate, data->redirect);
00347         break;
00348     }
00349     case MSG_DELETE_RULE:
00350     {
00351         delete_rule_t * data = (delete_rule_t *)msg->data;
00352 
00353         INSIST_ERR(msg->length == sizeof(delete_rule_t) + 
00354                         ntohs(data->rule_name_len));
00355 
00356         CLOG(LOG_INFO, "%s: Received a configuration update: Delete rule %s ",
00357                         __func__, data->rule_name);
00358 
00359         delete_rule(data->rule_name);
00360         break;
00361     }
00362     case MSG_CONF_RULE_MATCH_ADDR:
00363     {
00364         rule_addr_t * data = (rule_addr_t *)msg->data;
00365 
00366         INSIST_ERR(msg->length == sizeof(rule_addr_t) + 
00367                         ntohs(data->rule_name_len));
00368 
00369         CLOG(LOG_INFO, "%s: Received a configuration update: Add address to "
00370                         "rule %s ", __func__, data->rule_name);
00371 
00372         ia.s_addr = data->addr;
00373         CLOG(LOG_INFO, "%s: (%s with address: %s)", __func__,
00374                         data->rule_name, inet_ntoa(ia));
00375         
00376         ia.s_addr = data->mask;
00377         CLOG(LOG_INFO, "%s: (%s with mask: %s)", __func__,
00378                         data->rule_name, inet_ntoa(ia));
00379 
00380         add_address(data->rule_name, data->addr, data->mask);
00381 
00382         break;
00383     }
00384     case MSG_DELETE_RULE_MATCH_ADDR:
00385     {
00386         rule_addr_t * data = (rule_addr_t *)msg->data;
00387 
00388         INSIST_ERR(msg->length == sizeof(rule_addr_t) + 
00389                         ntohs(data->rule_name_len));
00390 
00391         CLOG(LOG_INFO, "%s: Received a configuration update: Delete address "
00392                         "from rule %s ", __func__, data->rule_name);
00393 
00394         ia.s_addr = data->addr;
00395         CLOG(LOG_INFO, "%s: (%s with address: %s)", __func__,
00396                         data->rule_name, inet_ntoa(ia));
00397         
00398         ia.s_addr = data->mask;
00399         CLOG(LOG_INFO, "%s: (%s with mask: %s)", __func__,
00400                         data->rule_name, inet_ntoa(ia));
00401 
00402         delete_address(data->rule_name, data->addr, data->mask);
00403 
00404         break;
00405     }
00406     default:
00407         CLOG(LOG_ERR, "%s: Received an unknown message type (%d) from the "
00408             "mgmt component", __func__, msg->subtype);
00409         return EFAIL;
00410     }
00411 
00412     return SUCCESS;
00413 }
00414 
00415 
00431 static void
00432 connect_mgmt(evContext ectx UNUSED,
00433             void * uap  UNUSED,
00434             struct timespec due UNUSED,
00435             struct timespec inter UNUSED)
00436 {
00437     pconn_client_params_t params;
00438 
00439     bzero(&params, sizeof(pconn_client_params_t));
00440 
00441     // setup the client args
00442     params.pconn_peer_info.ppi_peer_type = PCONN_PEER_TYPE_RE;
00443     params.pconn_port                    = MONITUBE_PORT_NUM;
00444     params.pconn_num_retries             = CONNECT_RETRIES;
00445     params.pconn_event_handler           = mgmt_client_connection;
00446 
00447     if(mgmt_client) {
00448         // Then this is the second time in a row this is called even though it
00449         // didn't fail. We haven't received an event like ESTABLISHED yet.
00450         pconn_client_close(mgmt_client);
00451     }
00452 
00453     // connect
00454     mgmt_client = pconn_client_connect_async(
00455                     &params, *ctx, mgmt_client_message, NULL);
00456 
00457     if(mgmt_client == NULL) {
00458         CLOG(LOG_ERR, "%s: Failed to initialize the pconn client connection "
00459             "to the mgmt component", __func__);
00460     }
00461 
00462     CLOG(LOG_INFO, "%s: Trying to connect to the mgmt component", __func__);
00463 }
00464 
00465 
00466 /*** GLOBAL/EXTERNAL Functions ***/
00467 
00468 
00472 void
00473 init_connections(void)
00474 {
00475     mgmt_client = NULL;
00476 
00477     evInitID(&mgmt_timer_id);
00478     msp_spinlock_init(&msgs_lock);
00479 
00480     // Connect to the MGMT component
00481     if(evSetTimer(*ctx, connect_mgmt, NULL, evNowTime(),
00482         evConsTime(RETRY_CONNECT, 0), &mgmt_timer_id)) {
00483 
00484         CLOG(LOG_EMERG, "%s: Failed to initialize a connect timer to connect "
00485             "to the MGMT component", __func__);
00486     }
00487 }
00488 
00489 
00493 void
00494 close_connections(void)
00495 {
00496     notification_msg_t * msg;
00497 
00498     if(mgmt_client) {
00499         pconn_client_close(mgmt_client);
00500         mgmt_client = NULL;
00501     }
00502 
00503     INSIST_ERR(msp_spinlock_lock(&msgs_lock) == MSP_OK);
00504 
00505     while((msg = TAILQ_FIRST(&notification_msgs)) != NULL) {
00506         TAILQ_REMOVE(&notification_msgs, msg, entries);
00507         free(msg->message);
00508         free(msg);
00509     }
00510 
00511     INSIST_ERR(msp_spinlock_unlock(&msgs_lock) == MSP_OK);
00512 }
00513 
00514 
00533 void
00534 notify_stat_update(in_addr_t flow_addr,
00535                    uint16_t flow_dport,
00536                    double mdi_df,
00537                    uint32_t mdi_mlr,
00538                    uint16_t ssid)
00539 {
00540     notification_msg_t * msg;
00541     void * doub;
00542     uint64_t out;
00543 
00544     if(mgmt_client == NULL) { // don't bother doing anything yet
00545         return;
00546     }
00547 
00548     msg = calloc(1, sizeof(notification_msg_t));
00549     INSIST_ERR(msg != NULL);
00550 
00551     msg->action = MSG_FLOW_STAT_UPDATE;
00552 
00553     msg->message = calloc(1, sizeof(flow_stat_t));
00554     msg->message->flow_addr = flow_addr;
00555     msg->message->flow_port = flow_dport;
00556     doub = &mdi_df; // play around double type in case of a truncating typecast
00557     out = *(uint64_t *)doub;
00558     msg->message->mdi_df = htonq(out);
00559     msg->message->mdi_mlr = htonl(mdi_mlr);
00560     msg->message->ss_id = htons(ssid);
00561 
00562     INSIST_ERR(msp_spinlock_lock(&msgs_lock) == MSP_OK);
00563 
00564     TAILQ_INSERT_TAIL(&notification_msgs, msg, entries);
00565 
00566     INSIST_ERR(msp_spinlock_unlock(&msgs_lock) == MSP_OK);
00567 
00568     process_notifications();
00569 }

2007-2009 Juniper Networks, Inc. All rights reserved. The information contained herein is confidential information of Juniper Networks, Inc., and may not be used, disclosed, distributed, modified, or copied without the prior written consent of Juniper Networks, Inc. in an express license. This information is subject to change by Juniper Networks, Inc. Juniper Networks, the Juniper Networks logo, and JUNOS are registered trademarks of Juniper Networks, Inc. in the United States and other countries. All other trademarks, service marks, registered trademarks, or registered service marks are the property of their respective owners.
Generated on Sun May 30 20:27:09 2010 for SDK Your Net Corporation Monitube2 IPTV Monitoring Example: monitube2-plugin 1.0 by Doxygen 1.5.1