dispatcher-role-5.4.cfg

### DISPATCHER ROLE ####

#!trydef KZ_DISPATCHER_PROBE_MODE 1
#!trydef DISPATCHER_ADD_SERVERS 1
#!trydef DISPATCHER_ADD_SECONDARY_IP 1
#!trydef DISPATCHER_SECONDARY_IP_GROUP 3
#!trydef DISPATCHER_ALG 0
#!trydef KZ_DISPATCHER_HASH_SIZE 8
#!trydef KZ_DISPATCHER_ADD_FLAGS 9
#!trydef KZ_DISPATCHER_PRIMARY_GROUP 1
#!trydef KZ_DISPATCHER_SECONDARY_GROUP 2
#!trydef KZ_DISPATCHER_CLASSIFY_GROUP 3
#!trydef KZ_DISPATCHER_ALTNET1_PRIMARY_GROUP 51
#!trydef KZ_DISPATCHER_ALTNET1_SECONDARY_GROUP 52
#!trydef KZ_DISPATCHER_ALTNET2_PRIMARY_GROUP 53
#!trydef KZ_DISPATCHER_ALTNET2_SECONDARY_GROUP 54
#!trydef KZ_DISPATCHER_PRESENCE_PRIMARY_GROUP 10
#!trydef KZ_DISPATCHER_PRESENCE_SECONDARY_GROUP 11
#!trydef KZ_DISPATCHER_REGISTRAR_PRIMARY_GROUP 20
#!trydef KZ_DISPATCHER_REGISTRAR_SECONDARY_GROUP 21
#!trydef KZ_DISPATCHER_MAX_RETRIES 2
#!trydef KZ_DISPATCHER_ROUTE_ASSOCIATED_MEDIA 1
#!trydef KZ_DISPATCHER_CLASSIFY_FLAGS 2
#!trydef KZ_DISPATCHER_PRINT_ROUTES 1

kazoo.dispatcher_auto_add = DISPATCHER_ADD_SERVERS descr "adds media servers reported by ecallmgr"
kazoo.dispatcher_add_secondary_ip = DISPATCHER_ADD_SECONDARY_IP descr "adds internal ip from media servers reported by ecallmgr"
kazoo.dispatcher_add_secondary_ip_group = DISPATCHER_SECONDARY_IP_GROUP descr "sets the group where to add internal ip from media servers reported by ecallmgr"
kazoo.dispatcher_algorithm = DISPATCHER_ALG descr "dispatcher algorithm to use"
kazoo.dispatcher_primary_group = KZ_DISPATCHER_PRIMARY_GROUP descr "dispatcher primary group"
kazoo.dispatcher_secondary_group = KZ_DISPATCHER_SECONDARY_GROUP descr "dispatcher secondary group"
kazoo.dispatcher_max_retries = KZ_DISPATCHER_MAX_RETRIES descr "max number of retries for media servers"
kazoo.dispatcher_route_to_associated_media = KZ_DISPATCHER_ROUTE_ASSOCIATED_MEDIA descr "routes to associated media for atxfer"
kazoo.dispatcher_classify_flags = KZ_DISPATCHER_CLASSIFY_FLAGS descr "dispatch classifier flags"
kazoo.dispatcher_print_routes = KZ_DISPATCHER_PRINT_ROUTES descr "should we log the selected routes"

####### Dispatcher module  ########
loadmodule "dispatcher.so"
modparam("dispatcher", "db_url", "KAZOO_DB_URL")
modparam("dispatcher", "flags", 2)
modparam("dispatcher", "use_default", 0)
modparam("dispatcher", "force_dst", 1)
modparam("dispatcher", "hash_pvar", "$avp(ds_grp)")
modparam("dispatcher", "setid_pvname", "$var(setid)")
modparam("dispatcher", "attrs_pvname", "$var(attrs)")
modparam("dispatcher", "ds_ping_method", "OPTIONS")
modparam("dispatcher", "ds_ping_interval", 10)
modparam("dispatcher", "ds_probing_threshold", 3)
modparam("dispatcher", "ds_probing_mode", KZ_DISPATCHER_PROBE_MODE)
modparam("dispatcher", "ds_ping_reply_codes", "501,403,404,400,200")
modparam("dispatcher", "ds_ping_from", "sip:sipcheck@MY_HOSTNAME")
modparam("dispatcher", "xavp_dst", "ds_dst")
modparam("dispatcher", "xavp_ctx", "ds_ctx")
modparam("dispatcher", "ds_hash_size", KZ_DISPATCHER_HASH_SIZE)

## Dispatcher Groups:
## 1 - Primary media servers
## 2 - Backup media servers
## 3 - Alternate media server IPs (used only for classification)
## 10 - Presence servers (if not locally handled)
## 20 - Registrar servers (if not locally handled)


modparam("rtimer", "timer", "name=dispatcher_reload;interval=20;mode=1;")
modparam("rtimer", "exec", "timer=dispatcher_reload;route=DISPATCHER_RELOAD")

####### Dispatcher Logic ########
route[CLASSIFY_SOURCE_DISPATCHER]
{
    route_if_exists("DISPATCHER_CUSTOM_NETWORK_CLASSIFY");

    if (!isflagset(FLAG_NETWORK_CLASSIFIED)) {
       if (is_myself("$ou")) {
           xlog("$var(log_request_level)", "original R-URI ($ou) is this proxy, treating as external sources\n");
       } else {
           $var(classify_dispatcher_flag) = $(sel(cfg_get.kazoo.dispatcher_classify_flags){s.int});
           if (ds_is_from_list(KZ_DISPATCHER_PRIMARY_GROUP, "$var(classify_dispatcher_flag)") ||
               ds_is_from_list(KZ_DISPATCHER_SECONDARY_GROUP, "$var(classify_dispatcher_flag)") ||
               ds_is_from_list(KZ_DISPATCHER_CLASSIFY_GROUP, "$var(classify_dispatcher_flag)") ||
               ds_is_from_list(KZ_DISPATCHER_ALTNET1_PRIMARY_GROUP, "$var(classify_dispatcher_flag)") ||
               ds_is_from_list(KZ_DISPATCHER_ALTNET1_SECONDARY_GROUP, "$var(classify_dispatcher_flag)") ||
               ds_is_from_list(KZ_DISPATCHER_ALTNET2_PRIMARY_GROUP, "$var(classify_dispatcher_flag)") ||
               ds_is_from_list(KZ_DISPATCHER_ALTNET2_SECONDARY_GROUP, "$var(classify_dispatcher_flag)") ||
               ds_is_from_list(KZ_DISPATCHER_PRESENCE_PRIMARY_GROUP, "$var(classify_dispatcher_flag)") ||
               ds_is_from_list(KZ_DISPATCHER_PRESENCE_SECONDARY_GROUP, "$var(classify_dispatcher_flag)") ||
               ds_is_from_list(KZ_DISPATCHER_REGISTRAR_PRIMARY_GROUP, "$var(classify_dispatcher_flag)") ||
               ds_is_from_list(KZ_DISPATCHER_REGISTRAR_SECONDARY_GROUP, "$var(classify_dispatcher_flag)") ||
               ($hdr(X-FS-Core-UUID) != $null && $sht(media=>$hdr(X-FS-Core-UUID)) != $null)) {
               xlog("$var(log_request_level)", "originated from internal sources\n");
               setflag(FLAG_INTERNALLY_SOURCED);
           } else {
               xlog("$var(log_request_level)", "originated from external sources\n");
           }
       }
    }
}

# Take the routes from dispatcher - hash over callid
# If prefered route defined, reorder the destinations
route[DISPATCHER_FIND_ROUTES]
{

    $var(ds_primary_group) = $(sel(cfg_get.kazoo.dispatcher_primary_group){s.int});
    $var(ds_backup_group) = $(sel(cfg_get.kazoo.dispatcher_secondary_group){s.int});
    
    #!ifndef PRESENCE_ROLE
    if (is_method("SUBSCRIBE")) {
        $var(ds_primary_group) = KZ_DISPATCHER_PRESENCE_PRIMARY_GROUP;
        $var(ds_backup_group)  = KZ_DISPATCHER_PRESENCE_SECONDARY_GROUP;
    }
    #!endif

    #!ifndef REGISTRAR_ROLE
    if (is_method("REGISTER")) {
        $var(ds_primary_group) = KZ_DISPATCHER_REGISTRAR_PRIMARY_GROUP;
        $var(ds_backup_group)  = KZ_DISPATCHER_REGISTRAR_SECONDARY_GROUP;
    }
    #!endif

    #!ifdef PRESENCE_ROLE
    route(PRESENCE_FAST_PICKUP_ATTEMPT);
    #!endif

    $var(ds_alg) = $sel(cfg_get.kazoo.dispatcher_algorithm);

    routes("DISPATCHER_SELECT_ROUTES");

    route(DISPATCHER_SELECT_ROUTES);
}

route[DISPATCHER_SELECT_ROUTES]
{
    $var(ds_group) = $var(ds_primary_group);
    
    if($var(ds_backup_group) != 0 && ds_list_exists("$var(ds_backup_group)")) {
        $var(ds_rule) = $_s($var(ds_primary_group)=$var(ds_alg);$var(ds_backup_group)=$var(ds_alg));
    } else {
        $var(ds_rule) = $_s($var(ds_primary_group)=$var(ds_alg));
    }

    ds_select_routes("$var(ds_rule)", "2");
    if ($xavp(ds_ctx=>cnt) == 0) {
        xlog("L_WARN", "end - no servers available in primary ($var(ds_primary_group)) or backup group ($var(ds_backup_group))\n");
        send_reply("480", "All servers busy");
        exit;
    }

    route(PRINT_ROUTES);

    $var(user_source) = $(ct{tobody.user}) + "@" + $si + ":" + $sp;
    if ($sht(associations=>$var(user_source)) != $null) {
        if($sel(cfg_get.kazoo.dispatcher_route_to_associated_media) == 1) {
            $var(prefered_route) = $sht(associations=>$var(user_source));
            xlog("L_INFO", "found association for contact uri $var(user_source)\n");
            route(DISPATCHER_PREFERRED_ROUTE);
        }
        $sht(associations=>$var(user_source)) = $null;
    }

    $avp(ds_group) = $xavp(ds_dst=>grp);
    $avp(ds_retries) = 0;
    ds_set_dst();

    routes("DISPATCHER_ROUTES_SELECTED");
}

route[DISPATCHER_PREFERRED_ROUTE]
{
    if(!ds_is_from_list(-1, 6, "$var(prefered_route)")) {
       xlog("L_INFO", "associated media server $var(prefered_route) is inactive, moving to $xavp(ds_dst=>uri)\n");
       return -1;
    }

    xlog("L_INFO", "re-ordering the dispatcher list to maintain association with $var(prefered_route)\n");

    $var(i) = 0;
    while($var(i) < $xavp(ds_ctx=>cnt)) {
       if($xavp(ds_dst[$var(i)]=>uri) == $var(prefered_route)) {
          $xavp(ds_dst[$var(i)]=>*) = $null;
       }
       $var(i) = $var(i) + 1;
    }

    $xavp(ds_dst=>uri) = $var(prefered_route);
    $xavp(ds_dst[0]=>grp) = $var(setid);
    $xavp(ds_dst[0]=>attrs) = $var(attrs);

    return 1;
}    

route[PRINT_ROUTES]
{
    if($sel(cfg_get.kazoo.dispatcher_print_routes) == 0) return;

    $var(i) = 0;
    while($xavp(ds_dst[$var(i)]=>uri) != $null) {
       xlog("L_INFO", "group $xavp(ds_dst[$var(i)]=>grp) => $xavp(ds_dst[$var(i)]=>uri) => zone $(xavp(ds_dst[$var(i)]=>attrs){param.value,zone})\n");
       $var(i) = $var(i) + 1;
    }

}

# Try next destinations in failure route
route[DISPATCHER_NEXT_ROUTE]
{
    if(!route_if_exists("DISPATCHER_CUSTOM_NEXT_ROUTE")) {
        if($avp(ds_retries) >= $sel(cfg_get.kazoo.dispatcher_max_retries)) return;

        $avp(ds_retries) = $avp(ds_retries) + 1;
        if(ds_next_dst()) {
            xlog("L_INFO", "routing call to next media server $du\n");

            setflag(FLAG_SKIP_NAT_CORRECTION);

            # relay the request to the next destination
            route(RELAY);

            exit;
        }
    }
}

event_route[dispatcher:dst-down]
{
    $ki = $uuid(g);
    xlog("L_WARNING", "destination $ru is down\n");
}

event_route[dispatcher:dst-up]
{
    $ki = $uuid(g);
    xlog("L_NOTICE", "destination $ru is up\n");
}

route[MEDIA_SERVER_UP_DISPATCHER]
{
   $var(check_media_server_ret) = 0;
   if($sel(cfg_get.kazoo.dispatcher_auto_add) == 1) {
       $var(SetId) = 1;
       if($var(Zone) != "MY_AMQP_ZONE") {
          $var(SetId) = 2;
       }

       $var(MediaUrlBack) = $var(MediaUrl);

       #!ifdef WITH_INTERNAL_LISTENER
       $var(dispatcher_socket) = $_s(socket=PROTO_INTERNAL;);
       $var(MediaUrl) = $_s($var(MediaUrl);transport=$def(INTERNAL_PROTO));
       #!else
       $var(dispatcher_socket) = "";
       #!endif
       $var(ping_headers) = $_s(X-Proxy-Core-UUID: $shv(kz_proxy_core_uuid));
       $var(flags) = KZ_DISPATCHER_ADD_FLAGS;
       $var(attrs) = $_s($var(dispatcher_socket)zone=$var(Zone);profile=$var(MediaProfile);duid=$(var(MediaUrl){s.corehash, MEDIA_SERVERS_HASH_SIZE});node=$var(MediaName));
       routes("DISPATCHER_MEDIA_CHECK");
       sql_query("exec", "KZQ_CHECK_MEDIA_SERVER_INSERT");
       if($sqlrows(exec) > 0) {
          $shv(dispatcher_reload) = 1;
          $var(check_media_server_ret) = 1;
       }

       if($sel(cfg_get.kazoo.dispatcher_add_secondary_ip) == 1) {
          if($var(MediaIP) != "" && $var(MediaIP) != $(var(MediaUrl){uri.host})) {
             #!ifdef WITH_INTERNAL_LISTENER
             $var(MediaUrl) = $_s($(var(MediaUrlBack){uri.scheme}):$var(MediaIP):$(var(MediaUrlBack){uri.port});transport=$def(INTERNAL_PROTO));
             #!else
             $var(MediaUrl) = $_s($(var(MediaUrlBack){uri.scheme}):$var(MediaIP):$(var(MediaUrlBack){uri.port}));
             #!endif
             $var(attrs) = $_s($var(dispatcher_socket)zone=$var(Zone);profile=$var(MediaProfile);duid=$(var(MediaUrl){s.corehash, MEDIA_SERVERS_HASH_SIZE});node=$var(MediaName));
             $var(SetId) = $sel(cfg_get.kazoo.dispatcher_add_secondary_ip_group);
             routes("DISPATCHER_MEDIA_CHECK");
             sql_query("exec", "KZQ_CHECK_MEDIA_SERVER_INSERT");
             if($sqlrows(exec) > 0) {
                $shv(dispatcher_reload) = 1;
                $var(check_media_server_ret) = 1;
             }
          }
       }

       $var(MediaUrl) = $var(MediaUrlBack);
   }
   return $var(check_media_server_ret);
}

route[DISPATCHER_RELOAD]
{
   $ki = $uuid(g);
   if($shv(dispatcher_reload) == 1) {
      xlog("L_WARNING", "reloading dispatcher table\n");
      ds_reload();
   };
   $shv(dispatcher_reload) = 0;  
}

route[DISPATCHER_STATUS]
{
    jsonrpc_exec('{"jsonrpc": "2.0", "method": "dispatcher.list", "id": 1}');
    $var(Sets) = $(jsonrpl(body){kz.json, result.NRSETS});
    $var(i) = 0;
    $var(ds_groups_json)="";
    $var(Sep1) = "";
    while($var(i) < $var(Sets)) {
       $var(Set) = $(jsonrpl(body){kz.json, result.RECORDS[$var(i)].SET});
       $var(SetCount) = $(var(Set){kz.json.count,TARGETS});
       $var(Sep2)="";
       $var(ds_group_json)="";
       $var(c) = 0;
       while($var(c) < $var(SetCount)) {
           $var(Dest) = $(var(Set){kz.json,TARGETS[$var(c)].DEST});
           $var(record) = $_s("$(var(Dest){kz.json,URI})" : {"destination" : "$(var(Dest){kz.json,URI})", "flags" : "$(var(Dest){kz.json,FLAGS})", "priority" : $(var(Dest){kz.json,PRIORITY}), "attrs" : "$(var(Dest){kz.json,ATTRS.BODY})"});
           $var(ds_group_json) = $var(ds_group_json) + $var(Sep2) + $var(record);
           $var(Sep2) = ",";
           $var(c) = $var(c) + 1;
       }
        $var(ds_groups_json) = $var(ds_groups_json) + $var(Sep1) + $_s("$(var(Set){kz.json,ID})" : { $var(ds_group_json) });
        $var(Sep1)=", ";
        $var(i) = $var(i) + 1;
    }
}

route[KZ_LOCAL_OPTIONS_INTERNAL]
{
    $var(uri) = $_s($(ru{uri.scheme}):$(ru{uri.host}):$(ru{uri.port}));
    if ($sht(media=>$var(uri)::uuid) != $null) {
        append_hf("X-Proxy-Core-UUID: $shv(kz_proxy_core_uuid)\r\n");
    }
}

route[ROUTE_TO_EXTERNAL_DISPATCHER_GROUP]
{
    if ($hdr(X-KAZOO-Dispatcher-Group) == $null)  {
       return;
    }

    $var(ds_primary_group) = $(hdr(X-KAZOO-Dispatcher-Group){s.int});
    $vn(dispatcher_route) = $hdr(X-KAZOO-Dispatcher-Route);

    xlog("L_INFO", "routing to group $hdr(X-KAZOO-Dispatcher-Group) : $hdr(X-KAZOO-Dispatcher-Route)\n");

    if ($vn(dispatcher_route) != $null)  {
        xlog("L_DEBUG", "running $_s($vn(dispatcher_route)_BEFORE_DISPATCHER)\n");
        route_if_exists($_s($vn(dispatcher_route)_BEFORE_DISPATCHER));
    }

    route(DISPATCHER_SELECT_ROUTES);

    if ($vn(dispatcher_route) != $null)  {
        xlog("L_DEBUG", "running $_s($vn(dispatcher_route)_AFTER_DISPATCHER)\n");
        route_if_exists($_s($vn(dispatcher_route)_AFTER_DISPATCHER));
        $xavp(cr=>dispatcher_route) = $vn(dispatcher_route);
    }

    setflag(FLAG_OUTBOUND_DISPATCHER);
}

route[ROUTE_TO_INTERNAL_DISPATCHER]
{
    route(DISPATCHER_FIND_ROUTES);
}


# vim: tabstop=4 softtabstop=4 shiftwidth=4 expandtab