tango-tsg-service-chaining-engine/platform/src/policy.cpp

#include <time.h>
#include <assert.h>
#include <arpa/inet.h>
#include <cjson/cJSON.h>
#include <MESA/maat.h>
#include <MESA/MESA_prof_load.h>

#include "global_metrics.h"
#include "health_check.h"
#include "packet.h"
#include "policy.h"
#include "utils.h"
#include "log.h"
#include "sce.h"
#include "utarray.h"

/******************************************************************************
 * Struct policy_enforcer
 ******************************************************************************/

#define EFFECTIVE_RANGE_MAX_SIZE 128

enum input_mode
{
    MAAT_INPUT_JSON = 0,
    MAAT_INPUT_REDIS = 1,
    MAAT_INPUT_FILE = 2,
};

struct policy_config
{
    enum input_mode input_mode;

    int log_level;
    int stat_switch;
    int perf_switch;
    int scan_detail;
    int deferred_load;
    int effect_interval_ms;

    char data_center[EFFECTIVE_RANGE_MAX_SIZE];
    char device_group[EFFECTIVE_RANGE_MAX_SIZE];
    char stat_file[2048];
    char table_info[2048];
    char accept_tags[2048];
    char accept_path[2048];
    char inc_cfg_dir[2048];
    char ful_cfg_dir[2048];
    char json_cfg_file[2048];
    char foreign_cont_dir[2048];

    int redis_db_idx;
    char redis_server[2048];
    char redis_port_range[2048];
    int max_chaining_size;
};

struct policy_enforcer
{
    struct policy_config config;
    struct maat *maat;

    int compile_table_id; // SERVICE_CHAINING_COMPILE           table id
    int sff_table_id;     // SERVICE_FUNCTION_FORWARDER_PROFILE table id
    int sf_table_id;      // SERVICE_FUNCTION_PROFILE           table id
};

/******************************************************************************
 * Struct chaining_param
 ******************************************************************************/

struct chaining_param
{
    uint64_t rule_id;
    int ref_cnt;
    int vsys_id;

    enum traffic_type traffic_type;
    int *sff_profile_ids;
    int sff_profile_ids_num;
};

/******************************************************************************
 * Struct sff_param
 ******************************************************************************/

enum failure_action
{
    FAILURE_ACTION_BYPASS = 1,
    FAILURE_ACTION_BLOCK = 2,
    FAILURE_ACTION_RE_DISPATCH = 3,
};

enum unavailable_action
{
    UNAVAILABLE_ACTION_BYPASSS = 1,
    UNAVAILABLE_ACTION_BLOCK = 2,
};

enum ldbc_localization
{
    LDBC_LOCALIZATION_NEARBY = 1,
    LDBC_LOCALIZATION_GLOBAL = 2,
};

struct exception
{
    enum failure_action fail_action;
    enum unavailable_action unavail_action;
    int health_service_func_lt;
};

struct load_balance
{
    enum ldbc_method method;
    enum ldbc_localization localiza;
};

struct sff_param
{
    int sff_profile_id;
    int sff_ref_cnt;

    enum forward_type sff_forward_type;
    struct load_balance sff_ldbc;
    struct exception sff_exception;

    int *sf_profile_ids;
    int sf_profile_ids_num;
};

/******************************************************************************
 * Struct sf_param
 ******************************************************************************/

enum admin_status
{
    ADMMIN_STATUS_ACTIVE = 1,
    ADMMIN_STATUS_INACTIVE = 2,
};

enum effective_type
{
    EFFECTIVE_TYPE_DEVICE_GROUP = 0x1,
    EFFECTIVE_TYPE_DATA_CENTER = 0x2,
};

struct effective_range
{
    enum effective_type type;
    char value[EFFECTIVE_RANGE_MAX_SIZE];
};

struct sf_param
{
    int sf_vsys_id;
    int sf_profile_id;
    int sf_ref_cnt;

    enum admin_status sf_admin_status;
    struct connectivity sf_connectivity;
    struct health_check sf_health_check;
    struct effective_range sf_effective_range;

    uint64_t health_check_session_id;
};

/******************************************************************************
 * Private API -- Utils
 ******************************************************************************/

static const char *effective_type_to_string(enum effective_type type)
{
    switch (type)
    {
    case EFFECTIVE_TYPE_DEVICE_GROUP:
        return "device_group";
    case EFFECTIVE_TYPE_DATA_CENTER:
        return "data_center";
    default:
        return "unknown";
    }
}

static const char *admin_status_to_string(enum admin_status admin_status)
{
    switch (admin_status)
    {
    case ADMMIN_STATUS_ACTIVE:
        return "active";
    case ADMMIN_STATUS_INACTIVE:
        return "inactive";
    default:
        return "unknown";
    }
}

// {"tags":[{"tag":"device_group","value":"group-xxg-9140"},{"tag":"data_center","value":"center-xxg-9140"}]}
static void parser_effective_range(const char *accept_tags, char *data_center, char *device_group)
{
    cJSON *json;
    cJSON *tags;
    cJSON *elem;
    cJSON *item_key;
    cJSON *item_val;

    json = cJSON_Parse(accept_tags);
    if (json == NULL)
    {
        LOG_ERROR("%s: unexpected maat config: (invalid accept_tags format) %s", LOG_TAG_POLICY, accept_tags);
        goto error_out;
    }

    tags = cJSON_GetObjectItem(json, "tags");
    if (!tags || !cJSON_IsArray(tags) || !cJSON_GetArraySize(tags))
    {
        LOG_ERROR("%s: unexpected maat config: (invalid accept_tags->tags format) %s", LOG_TAG_POLICY, accept_tags);
        goto error_out;
    }

    cJSON_ArrayForEach(elem, tags)
    {
        item_key = cJSON_GetObjectItem(elem, "tag");
        if (!item_key || !cJSON_IsString(item_key))
        {
            LOG_ERROR("%s: unexpected maat config: (invalid accept_tags->tags->tag format) %s", LOG_TAG_POLICY, accept_tags);
            continue;
        }

        item_val = cJSON_GetObjectItem(elem, "value");
        if (!item_val || !cJSON_IsString(item_val))
        {
            LOG_ERROR("%s: unexpected maat config: (invalid accept_tags->tags->value format) %s", LOG_TAG_POLICY, accept_tags);
            continue;
        }

        if (strcasecmp(item_key->valuestring, "device_group") == 0)
        {
            memset(device_group, 0, EFFECTIVE_RANGE_MAX_SIZE);
            memcpy(device_group, item_val->valuestring, MIN(strlen(item_val->valuestring), EFFECTIVE_RANGE_MAX_SIZE));
        }

        if (strcasecmp(item_key->valuestring, "data_center") == 0)
        {
            memset(data_center, 0, EFFECTIVE_RANGE_MAX_SIZE);
            memcpy(data_center, item_val->valuestring, MIN(strlen(item_val->valuestring), EFFECTIVE_RANGE_MAX_SIZE));
        }
    }

error_out:
    if (json)
    {
        cJSON_Delete(json);
        json = NULL;
    }
}

static void policy_enforcer_config(const char *profile, struct policy_config *config)
{
    MESA_load_profile_int_def(profile, "MAAT", "input_mode", (int *)&(config->input_mode), MAAT_INPUT_REDIS);
    // LOG_LEVEL_TRACE = 0; LOG_LEVEL_DEBUG = 1; LOG_LEVEL_INFO  = 2;
    // LOG_LEVEL_WARN  = 3; LOG_LEVEL_ERROR = 4; LOG_LEVEL_FATAL = 5;
    MESA_load_profile_int_def(profile, "MAAT", "log_level", &(config->log_level), 5);
    MESA_load_profile_int_def(profile, "MAAT", "stat_switch", &(config->stat_switch), 1);
    MESA_load_profile_int_def(profile, "MAAT", "perf_switch", &(config->perf_switch), 1);
    MESA_load_profile_int_def(profile, "MAAT", "scan_detail", &(config->scan_detail), 0);
    MESA_load_profile_int_def(profile, "MAAT", "deferred_load", &(config->deferred_load), 0);
    MESA_load_profile_int_def(profile, "MAAT", "effect_interval_ms", &(config->effect_interval_ms), 1000);

    MESA_load_profile_string_def(profile, "MAAT", "stat_file", config->stat_file, sizeof(config->stat_file), "log/maat.fs2");
    MESA_load_profile_string_def(profile, "MAAT", "table_info", config->table_info, sizeof(config->table_info), "resource/table_info.conf");
    MESA_load_profile_string_def(profile, "MAAT", "accept_path", config->accept_path, sizeof(config->accept_path), "/opt/tsg/etc/tsg_device_tag.json");
    MESA_load_profile_string_def(profile, "MAAT", "inc_cfg_dir", config->inc_cfg_dir, sizeof(config->inc_cfg_dir), "resource/inc/");
    MESA_load_profile_string_def(profile, "MAAT", "ful_cfg_dir", config->ful_cfg_dir, sizeof(config->ful_cfg_dir), "resource/ful/");
    MESA_load_profile_string_def(profile, "MAAT", "json_cfg_file", config->json_cfg_file, sizeof(config->json_cfg_file), "resource/sce.json");
    MESA_load_profile_string_def(profile, "MAAT", "foreign_cont_dir", config->foreign_cont_dir, sizeof(config->foreign_cont_dir), "resource/sce_files");

    MESA_load_profile_int_def(profile, "MAAT", "redis_db_idx", &(config->redis_db_idx), 0);
    MESA_load_profile_string_def(profile, "MAAT", "redis_server", config->redis_server, sizeof(config->redis_server), "127.0.0.1");
    MESA_load_profile_string_def(profile, "MAAT", "redis_port_range", config->redis_port_range, sizeof(config->redis_server), "6379");
    MESA_load_profile_int_def(profile, "MAAT", "max_chaining_size", &(config->max_chaining_size), 32);

    if (strlen(config->accept_path))
    {
        MESA_load_profile_string_def(config->accept_path, "MAAT", "accept_tags", config->accept_tags, sizeof(config->accept_tags), "{\"tags\":[{\"tag\":\"device_id\",\"value\":\"device_1\"}]}");
        parser_effective_range(config->accept_tags, config->data_center, config->device_group);
    }

    LOG_DEBUG("%s: MAAT->input_mode         : %s", LOG_TAG_POLICY, (config->input_mode == MAAT_INPUT_REDIS ? "redis" : (config->input_mode == MAAT_INPUT_JSON ? "json" : (config->input_mode == MAAT_INPUT_FILE ? "file" : "unknown"))));
    LOG_DEBUG("%s: MAAT->log_level          : %d", LOG_TAG_POLICY, config->log_level);
    LOG_DEBUG("%s: MAAT->stat_switch        : %d", LOG_TAG_POLICY, config->stat_switch);
    LOG_DEBUG("%s: MAAT->perf_switch        : %d", LOG_TAG_POLICY, config->perf_switch);
    LOG_DEBUG("%s: MAAT->scan_detail        : %d", LOG_TAG_POLICY, config->scan_detail);
    LOG_DEBUG("%s: MAAT->deferred_load      : %d", LOG_TAG_POLICY, config->deferred_load);
    LOG_DEBUG("%s: MAAT->effect_interval_ms : %d", LOG_TAG_POLICY, config->effect_interval_ms);

    LOG_DEBUG("%s: MAAT->stat_file          : %s", LOG_TAG_POLICY, config->stat_file);
    LOG_DEBUG("%s: MAAT->table_info         : %s", LOG_TAG_POLICY, config->table_info);
    LOG_DEBUG("%s: MAAT->accept_path        : %s", LOG_TAG_POLICY, config->accept_path);
    LOG_DEBUG("%s: MAAT->accept_tags        : %s", LOG_TAG_POLICY, config->accept_tags);
    LOG_DEBUG("%s: MAAT->device_group       : %s", LOG_TAG_POLICY, config->device_group);
    LOG_DEBUG("%s: MAAT->data_center        : %s", LOG_TAG_POLICY, config->data_center);
    LOG_DEBUG("%s: MAAT->inc_cfg_dir        : %s", LOG_TAG_POLICY, config->inc_cfg_dir);
    LOG_DEBUG("%s: MAAT->ful_cfg_dir        : %s", LOG_TAG_POLICY, config->ful_cfg_dir);
    LOG_DEBUG("%s: MAAT->json_cfg_file      : %s", LOG_TAG_POLICY, config->json_cfg_file);
    LOG_DEBUG("%s: MAAT->foreign_cont_dir   : %s", LOG_TAG_POLICY, config->foreign_cont_dir);

    LOG_DEBUG("%s: MAAT->redis_db_idx       : %d", LOG_TAG_POLICY, config->redis_db_idx);
    LOG_DEBUG("%s: MAAT->redis_server       : %s", LOG_TAG_POLICY, config->redis_server);
    LOG_DEBUG("%s: MAAT->redis_port_range   : %s", LOG_TAG_POLICY, config->redis_port_range);
    LOG_DEBUG("%s: MAAT->max_chaining_size  : %d", LOG_TAG_POLICY, config->max_chaining_size);
}

/******************************************************************************
 * Private API -- MAAT Callback
 ******************************************************************************/

static void chaining_param_new_cb(const char *table_name, int table_id, const char *key, const char *table_line, void **ad, long argl, void *argp)
{
    int iter = 0;
    cJSON *json = NULL;
    cJSON *item = NULL;
    cJSON *element = NULL;
    size_t user_region_offset = 0;
    size_t user_region_len = 0;
    struct chaining_param *param = NULL;

    if (maat_helper_read_column(table_line, 7, &user_region_offset, &user_region_len) < 0)
    {
        LOG_ERROR("%s: unexpected chaining rule: (invalid user region) %s", LOG_TAG_POLICY, table_line);
        return;
    }

    char *json_str = (char *)calloc(user_region_len + 1, sizeof(char));
    memcpy(json_str, table_line + user_region_offset, user_region_len);
    json = cJSON_Parse(json_str);
    if (json == NULL)
    {
        LOG_ERROR("%s: unexpected chaining rule: (invalid json format) %s", LOG_TAG_POLICY, table_line);
        goto error_out;
    }

    param = (struct chaining_param *)calloc(1, sizeof(struct chaining_param));
    param->rule_id = atoll(key);
    param->ref_cnt = 1;

    // vsys_id
    item = cJSON_GetObjectItem(json, "vsys_id");
    if (!item || !cJSON_IsNumber(item))
    {
        LOG_ERROR("%s: unexpected chaining rule: (invalid vsys_id param) %s", LOG_TAG_POLICY, table_line);
        goto error_out;
    }
    param->vsys_id = item->valueint;
    LOG_DEBUG("%s: parse chaining rule: %lu, vsys_id: %d", LOG_TAG_POLICY, param->rule_id, param->vsys_id);

    // targeted_traffic
    item = cJSON_GetObjectItem(json, "targeted_traffic");
    if (!item || !cJSON_IsString(item))
    {
        LOG_ERROR("%s: unexpected chaining rule: (invalid targeted_traffic param) %s", LOG_TAG_POLICY, table_line);
        goto error_out;
    }
    if (strcasecmp(item->valuestring, "raw") == 0)
    {
        param->traffic_type = TRAFFIC_TYPE_RAW;
    }
    else if (strcasecmp(item->valuestring, "decrypted") == 0)
    {
        param->traffic_type = TRAFFIC_TYPE_DECRYPTED;
    }
    else
    {
        LOG_ERROR("%s: unexpected chaining rule: (invalid targeted_traffic param) %s", LOG_TAG_POLICY, table_line);
        goto error_out;
    }
    LOG_DEBUG("%s: parse chaining rule: %lu, targeted_traffic: %s", LOG_TAG_POLICY, param->rule_id, traffic_type_to_string(param->traffic_type));

    // sff_profiles
    item = cJSON_GetObjectItem(json, "sff_profiles");
    if (!item || !cJSON_IsArray(item) || !cJSON_GetArraySize(item))
    {
        LOG_ERROR("%s: unexpected chaining rule: (invalid sff_profiles param) %s", LOG_TAG_POLICY, table_line);
        goto error_out;
    }
    param->sff_profile_ids_num = cJSON_GetArraySize(item);
    param->sff_profile_ids = (int *)calloc(param->sff_profile_ids_num, sizeof(int));
    cJSON_ArrayForEach(element, item)
    {
        if (!cJSON_IsNumber(element))
        {
            LOG_ERROR("%s: unexpected chaining rule: (invalid sff_profiles param) %s", LOG_TAG_POLICY, table_line);
            continue;
        }

        LOG_DEBUG("%s: parse chaining rule: %lu, sff_profiles[%d/%d]: %d", LOG_TAG_POLICY, param->rule_id, iter, param->sff_profile_ids_num, element->valueint);
        param->sff_profile_ids[iter] = element->valueint;
        iter++;
    }

    *ad = param;
    LOG_INFO("%s: Add chaining rule: %lu", LOG_TAG_POLICY, param->rule_id);

    cJSON_Delete(json);
    free(json_str);
    return;

error_out:
    if (json)
    {
        cJSON_Delete(json);
        json = NULL;
    }

    if (json_str)
    {
        free(json_str);
        json_str = NULL;
    }

    if (param)
    {
        if (param->sff_profile_ids)
        {
            free(param->sff_profile_ids);
            param->sff_profile_ids = NULL;
        }
        free(param);
        param = NULL;
    }
}

static void chaining_param_free_cb(int table_id, void **ad, long argl, void *argp)
{
    struct chaining_param *param = (struct chaining_param *)*ad;
    if (param == NULL)
    {
        return;
    }

    if ((__sync_sub_and_fetch(&param->ref_cnt, 1) == 0))
    {
        LOG_INFO("%s: Del chaining rule: %lu", LOG_TAG_POLICY, param->rule_id);
        if (param->sff_profile_ids)
        {
            free(param->sff_profile_ids);
            param->sff_profile_ids = NULL;
        }
        free(param);
        param = NULL;

        *ad = NULL;
    }
}

static void chaining_param_dup_cb(int table_id, void **to, void **from, long argl, void *argp)
{
    struct chaining_param *param = (struct chaining_param *)*from;
    if (param)
    {
        __sync_add_and_fetch(&(param->ref_cnt), 1);
        *to = param;
    }
    else
    {
        *to = NULL;
    }
}

static void chaining_param_free(struct chaining_param *param)
{
    chaining_param_free_cb(0, (void **)&param, 0, NULL);
}

static void sff_param_new_cb(const char *table_name, int table_id, const char *key, const char *table_line, void **ad, long argl, void *argp)
{
    int iter = 0;
    struct sff_param *param = NULL;
    cJSON *root1 = NULL;
    cJSON *root2 = NULL;
    cJSON *item = NULL;

    int profile_id = 0;
    int type = 0;
    char load_balance_method[32] = {0};
    char load_balance_localization[8] = {0};
    char failure_action[16] = {0};
    char unavailability_action[64] = {0};
    char service_func_profiles[128] = {0};
    int is_valid = 0;

    if (sscanf(table_line, "%d\t%d\t%s\t%s\t%s\t%s\t%s\t%d",
               &profile_id, &type,
               load_balance_method, load_balance_localization, failure_action, unavailability_action, service_func_profiles,
               &is_valid) != 8)
    {
        LOG_ERROR("%s: unexpected sff profile: %s", LOG_TAG_POLICY, table_line);
        return;
    }

    param = (struct sff_param *)calloc(1, sizeof(struct sff_param));
    param->sff_profile_id = atoi(key);
    param->sff_ref_cnt = 1;

    // type
    switch (type)
    {
    case 1:
        param->sff_forward_type = FORWARD_TYPE_STEERING;
        break;
    case 2:
        param->sff_forward_type = FORWARD_TYPE_MIRRORING;
        break;
    default:
        LOG_ERROR("%s: unexpected sff profile: (invalid type param) %s", LOG_TAG_POLICY, table_line);
        goto error_out;
    }
    LOG_DEBUG("%s: parse sff profile: %d, type: %s", LOG_TAG_POLICY, param->sff_profile_id, forward_type_to_string(param->sff_forward_type));

    // load_balance_method
    if (0 == strcasecmp(load_balance_method, "hash-int-ip"))
    {
        param->sff_ldbc.method = LDBC_METHOD_HASH_INT_IP;
    }
    else if (0 == strcasecmp(load_balance_method, "hash-ext-ip"))
    {
        param->sff_ldbc.method = LDBC_METHOD_HASH_EXT_IP;
    }
    else if (0 == strcasecmp(load_balance_method, "hash-int-ip-and-ext-ip"))
    {
        param->sff_ldbc.method = LDBC_METHOD_HASH_INT_IP_AND_EXT_IP;
    }
    else if (0 == strcasecmp(load_balance_method, "hash-innermost-int-ip"))
    {
        param->sff_ldbc.method = LDBC_METHOD_HASH_INNERMOST_INT_IP;
    }
    else
    {
        // not support LDBC_METHOD_HASH_INNERMOST_EXT_IP
        LOG_ERROR("%s: unexpected sff profile: (invalid load_balance_method param) %s", LOG_TAG_POLICY, table_line);
        goto error_out;
    }
    LOG_DEBUG("%s: parse sff profile: %d, load_balance_method: %s", LOG_TAG_POLICY, param->sff_profile_id, load_balance_method);

    // load_balance_localization
    if (0 == strcasecmp(load_balance_localization, "nearby"))
    {
        param->sff_ldbc.localiza = LDBC_LOCALIZATION_NEARBY;
    }
    else if (0 == strcasecmp(load_balance_localization, "global"))
    {
        param->sff_ldbc.localiza = LDBC_LOCALIZATION_GLOBAL;
    }
    else
    {
        LOG_ERROR("%s: unexpected sff profile: (invalid load_balance_localization param) %s", LOG_TAG_POLICY, table_line);
        goto error_out;
    }
    LOG_DEBUG("%s: parse sff profile: %d, load_balance_localization: %s", LOG_TAG_POLICY, param->sff_profile_id, load_balance_localization);

    // failure_action
    if (0 == strcasecmp(failure_action, "bypass"))
    {
        param->sff_exception.fail_action = FAILURE_ACTION_BYPASS;
    }
    else if (0 == strcasecmp(failure_action, "block"))
    {
        param->sff_exception.fail_action = FAILURE_ACTION_BLOCK;
    }
    else if (0 == strcasecmp(failure_action, "re-dispatch"))
    {
        param->sff_exception.fail_action = FAILURE_ACTION_RE_DISPATCH;
    }
    else
    {
        LOG_ERROR("%s: unexpected sff profile: (invalid failure_action param) %s", LOG_TAG_POLICY, table_line);
        goto error_out;
    }
    LOG_DEBUG("%s: parse sff profile: %d, failure_action: %s", LOG_TAG_POLICY, param->sff_profile_id, failure_action);

    // unavailability_action
    if (param->sff_exception.fail_action == FAILURE_ACTION_RE_DISPATCH)
    {
        root1 = cJSON_Parse(unavailability_action);
        if (root1 == NULL)
        {
            LOG_ERROR("%s: unexpected sff profile: (invalid unavailability_action param) %s", LOG_TAG_POLICY, table_line);
            goto error_out;
        }
        item = cJSON_GetObjectItem(root1, "action");
        if (!item || !cJSON_IsString(item))
        {
            LOG_ERROR("%s: unexpected chaining rule: (invalid unavailability_action->action param) %s", LOG_TAG_POLICY, table_line);
            goto error_out;
        }
        if (0 == strcasecmp(item->valuestring, "bypass"))
        {
            param->sff_exception.unavail_action = UNAVAILABLE_ACTION_BYPASSS;
        }
        else if (0 == strcasecmp(item->valuestring, "block"))
        {
            param->sff_exception.unavail_action = UNAVAILABLE_ACTION_BLOCK;
        }
        else
        {
            LOG_ERROR("%s: unexpected chaining rule: (invalid unavailability_action->action param) %s", LOG_TAG_POLICY, table_line);
            goto error_out;
        }
        LOG_DEBUG("%s: parse sff profile: %d, unavailability_action->action: %s", LOG_TAG_POLICY, param->sff_profile_id, item->valuestring);

        item = cJSON_GetObjectItem(root1, "health_service_func_lt");
        if (item && cJSON_IsNumber(item))
        {
            param->sff_exception.health_service_func_lt = item->valueint;
            LOG_DEBUG("%s: parse sff profile: %d, unavailability_action->health_service_func_lt: %d", LOG_TAG_POLICY, param->sff_profile_id, item->valueint);
        }
    }

    // service_func_profiles
    root2 = cJSON_Parse(service_func_profiles);
    if (root2 == NULL || !cJSON_IsArray(root2) || !cJSON_GetArraySize(root2))
    {
        LOG_ERROR("%s: unexpected sff profile: (invalid service_func_profiles param) %s", LOG_TAG_POLICY, table_line);
        return;
    }
    param->sf_profile_ids_num = cJSON_GetArraySize(root2);
    param->sf_profile_ids = (int *)calloc(param->sf_profile_ids_num, sizeof(int));
    cJSON_ArrayForEach(item, root2)
    {
        if (!cJSON_IsNumber(item))
        {
            LOG_ERROR("%s: unexpected sff profile: (invalid service_func_profiles param) %s", LOG_TAG_POLICY, table_line);
            continue;
        }
        LOG_DEBUG("%s: parse sff profile: %d, service_func_profiles[%d/%d] = %d", LOG_TAG_POLICY, param->sff_profile_id, iter, param->sf_profile_ids_num, item->valueint);
        param->sf_profile_ids[iter] = item->valueint;
        iter++;
    }

    *ad = param;
    LOG_INFO("%s: Add sff profile: %d", LOG_TAG_POLICY, param->sff_profile_id);

    cJSON_Delete(root1);
    cJSON_Delete(root2);
    return;

error_out:
    if (root1)
    {
        cJSON_Delete(root1);
        root1 = NULL;
    }

    if (root2)
    {
        cJSON_Delete(root2);
        root2 = NULL;
    }

    if (param)
    {
        if (param->sf_profile_ids)
        {
            free(param->sf_profile_ids);
            param->sf_profile_ids = NULL;
        }
        free(param);
        param = NULL;
    }
}

static void sff_param_free_cb(int table_id, void **ad, long argl, void *argp)
{
    struct sff_param *param = (struct sff_param *)*ad;
    if (param == NULL)
    {
        return;
    }

    if ((__sync_sub_and_fetch(&param->sff_ref_cnt, 1) == 0))
    {
        LOG_INFO("%s: Del sff profile: %d", LOG_TAG_POLICY, param->sff_profile_id);
        if (param->sf_profile_ids)
        {
            free(param->sf_profile_ids);
            param->sf_profile_ids = NULL;
        }
        free(param);
        param = NULL;

        *ad = NULL;
    }
}

static void sff_param_dup_cb(int table_id, void **to, void **from, long argl, void *argp)
{
    struct sff_param *param = (struct sff_param *)*from;
    if (param)
    {
        __sync_add_and_fetch(&(param->sff_ref_cnt), 1);
        *to = param;
    }
    else
    {
        *to = NULL;
    }
}

static void sff_param_free(struct sff_param *param)
{
    sff_param_free_cb(0, (void **)&param, 0, NULL);
}

static void sf_param_new_cb(const char *table_name, int table_id, const char *key, const char *table_line, void **ad, long argl, void *argp)
{
    struct sf_param *param = NULL;
    cJSON *root0 = NULL;
    cJSON *root1 = NULL;
    cJSON *root2 = NULL;
    cJSON *item = NULL;

    int vsys_id = 0;
    int is_valid = 0;
    int profile_id = 0;
    int admin_status = 0;
    char connectivity[128] = {0};
    char health_check[128] = {0};
    char device_group[EFFECTIVE_RANGE_MAX_SIZE] = {0};

    if (sscanf(table_line, "%d\t%s\t%d\t%s\t%s\t%d\t%d",
               &profile_id, device_group, &admin_status, connectivity, health_check, &vsys_id, &is_valid) != 7)
    {
        LOG_ERROR("%s: unexpected sf profile: %s", LOG_TAG_POLICY, table_line);
        return;
    }

    param = (struct sf_param *)calloc(1, sizeof(struct sf_param));
    param->sf_vsys_id = vsys_id;
    param->sf_profile_id = atoi(key);
    param->sf_ref_cnt = 1;

    // device_group
    root0 = cJSON_Parse(device_group);
    if (root0 == NULL)
    {
        LOG_ERROR("%s: unexpected sf profile: (invalid device_group param) %s", LOG_TAG_POLICY, table_line);
        goto error_out;
    }
    item = cJSON_GetObjectItem(root0, "tag");
    if (!item || !cJSON_IsString(item))
    {
        LOG_ERROR("%s: unexpected sf profile: (invalid device_group->tag param) %s", LOG_TAG_POLICY, table_line);
        goto error_out;
    }
    if (0 == strcasecmp(item->valuestring, "device_group"))
    {
        param->sf_effective_range.type = EFFECTIVE_TYPE_DEVICE_GROUP;
    }
    else if (0 == strcasecmp(item->valuestring, "data_center"))
    {
        param->sf_effective_range.type = EFFECTIVE_TYPE_DATA_CENTER;
    }
    else
    {
        LOG_ERROR("%s: unexpected sf profile: (invalid device_group->tag param) %s", LOG_TAG_POLICY, table_line);
        goto error_out;
    }
    item = cJSON_GetObjectItem(root0, "value");
    if (!item || !cJSON_IsString(item))
    {
        LOG_ERROR("%s: unexpected sf profile: (invalid device_group->value param) %s", LOG_TAG_POLICY, table_line);
        goto error_out;
    }
    memcpy(param->sf_effective_range.value, item->valuestring, MIN(strlen(item->valuestring), EFFECTIVE_RANGE_MAX_SIZE));
    LOG_DEBUG("%s: parse sf profile: %d, device_group->tag: %s, device_group->value: %s", LOG_TAG_POLICY, param->sf_profile_id, effective_type_to_string(param->sf_effective_range.type), param->sf_effective_range.value);

    // admin_status
    switch (admin_status)
    {
    case 1:
        param->sf_admin_status = ADMMIN_STATUS_ACTIVE;
        break;
    case 0:
        param->sf_admin_status = ADMMIN_STATUS_INACTIVE;
        break;
    default:
        LOG_ERROR("%s: unexpected sf profile: (invalid admin_status param) %s", LOG_TAG_POLICY, table_line);
        goto error_out;
    }
    LOG_DEBUG("%s: parse sf profile: %d, admin_status: %s", LOG_TAG_POLICY, param->sf_profile_id, admin_status_to_string(param->sf_admin_status));

    // connectivity
    root1 = cJSON_Parse(connectivity);
    if (root1 == NULL)
    {
        LOG_ERROR("%s: unexpected sf profile: (invalid connectivity param) %s", LOG_TAG_POLICY, table_line);
        goto error_out;
    }

    item = cJSON_GetObjectItem(root1, "method");
    if (!item || !cJSON_IsString(item))
    {
        LOG_ERROR("%s: unexpected sf profile: (invalid connectivity->method param) %s", LOG_TAG_POLICY, table_line);
        goto error_out;
    }
    if (0 == strcasecmp(item->valuestring, "layer2_switch"))
    {
        param->sf_connectivity.method = ENCAPSULATE_METHOD_LAYER2_SWITCH;
    }
    else if (0 == strcasecmp(item->valuestring, "layer3_switch"))
    {
        param->sf_connectivity.method = ENCAPSULATE_METHOD_LAYER3_SWITCH;
    }
    else if (0 == strcasecmp(item->valuestring, "vxlan_g"))
    {
        param->sf_connectivity.method = ENCAPSULATE_METHOD_VXLAN_G;
    }
    else
    {
        LOG_ERROR("%s: unexpected sf profile: (invalid connectivity->method param) %s", LOG_TAG_POLICY, table_line);
        goto error_out;
    }
    LOG_DEBUG("%s: parse sf profile: %d, connectivity->method: %s", LOG_TAG_POLICY, param->sf_profile_id, encapsulate_method_to_string(param->sf_connectivity.method));

    if (param->sf_connectivity.method == ENCAPSULATE_METHOD_LAYER2_SWITCH || param->sf_connectivity.method == ENCAPSULATE_METHOD_LAYER3_SWITCH)
    {
        item = cJSON_GetObjectItem(root1, "int_vlan_tag");
        if (!item || !cJSON_IsNumber(item))
        {
            LOG_ERROR("%s: unexpected sf profile: (invalid connectivity->int_vlan_tag param) %s", LOG_TAG_POLICY, table_line);
            goto error_out;
        }
        param->sf_connectivity.int_vlan_tag = item->valueint;
        LOG_DEBUG("%s: parse sf profile: %d, connectivity->int_vlan_tag: %d", LOG_TAG_POLICY, param->sf_profile_id, item->valueint);

        item = cJSON_GetObjectItem(root1, "ext_vlan_tag");
        if (!item || !cJSON_IsNumber(item))
        {
            LOG_ERROR("%s: unexpected sf profile: (invalid connectivity->ext_vlan_tag param) %s", LOG_TAG_POLICY, table_line);
            goto error_out;
        }
        param->sf_connectivity.ext_vlan_tag = item->valueint;
        LOG_DEBUG("%s: parse sf profile: %d, connectivity->ext_vlan_tag: %d", LOG_TAG_POLICY, param->sf_profile_id, item->valueint);
    }
    else if (param->sf_connectivity.method == ENCAPSULATE_METHOD_VXLAN_G)
    {
        item = cJSON_GetObjectItem(root1, "dest_ip");
        if (!item || !cJSON_IsString(item))
        {
            LOG_ERROR("%s: unexpected sf profile: (invalid connectivity->dest_ip param) %s", LOG_TAG_POLICY, table_line);
            goto error_out;
        }
        memcpy(param->sf_connectivity.dest_ip, item->valuestring, strlen(item->valuestring));
        LOG_DEBUG("%s: parse sf profile: %d, connectivity->dest_ip: %s", LOG_TAG_POLICY, param->sf_profile_id, item->valuestring);
    }

    // health_check
    root2 = cJSON_Parse(health_check);
    if (root2 == NULL)
    {
        LOG_ERROR("%s: unexpected sf profile: (invalid health_check param) %s", LOG_TAG_POLICY, table_line);
        goto error_out;
    }
    item = cJSON_GetObjectItem(root2, "method");
    if (!item || !cJSON_IsString(item))
    {
        LOG_ERROR("%s: unexpected sf profile: (invalid health_check->method param) %s", LOG_TAG_POLICY, table_line);
        goto error_out;
    }
    if (0 == strcasecmp(item->valuestring, "none"))
    {
        param->sf_health_check.method = HEALTH_CHECK_METHOD_NONE;
    }
    else if (0 == strcasecmp(item->valuestring, "in_band_bfd"))
    {
        param->sf_health_check.method = HEALTH_CHECK_METHOD_IN_BAND_BFD;
    }
    else if (0 == strcasecmp(item->valuestring, "bfd"))
    {
        param->sf_health_check.method = HEALTH_CHECK_METHOD_BFD;
    }
    else if (0 == strcasecmp(item->valuestring, "http"))
    {
        param->sf_health_check.method = HEALTH_CHECK_METHOD_HTTP;
    }
    else
    {
        LOG_ERROR("%s: unexpected sf profile: (invalid health_check->method param) %s", LOG_TAG_POLICY, table_line);
        goto error_out;
    }
    LOG_DEBUG("%s: parse sf profile: %d, health_check->method: %s", LOG_TAG_POLICY, param->sf_profile_id, item->valuestring);

    if ((param->sf_health_check.method == HEALTH_CHECK_METHOD_BFD && param->sf_connectivity.method == ENCAPSULATE_METHOD_VXLAN_G) ||
        (param->sf_health_check.method == HEALTH_CHECK_METHOD_NONE && param->sf_connectivity.method == ENCAPSULATE_METHOD_VXLAN_G))
    {
        memcpy(param->sf_health_check.address, param->sf_connectivity.dest_ip, strlen(param->sf_connectivity.dest_ip));
    }

    if (param->sf_health_check.method == HEALTH_CHECK_METHOD_HTTP)
    {
        item = cJSON_GetObjectItem(root2, "url");
        if (!item || !cJSON_IsString(item))
        {
            LOG_ERROR("%s: unexpected sf profile: (invalid health_check->url param) %s", LOG_TAG_POLICY, table_line);
            goto error_out;
        }
        memcpy(param->sf_health_check.url, item->valuestring, strlen(item->valuestring));
        LOG_DEBUG("%s: parse sf profile: %d, health_check->url: %s", LOG_TAG_POLICY, param->sf_profile_id, item->valuestring);
    }

    if (param->sf_health_check.method == HEALTH_CHECK_METHOD_HTTP || param->sf_health_check.method == HEALTH_CHECK_METHOD_BFD || param->sf_health_check.method == HEALTH_CHECK_METHOD_IN_BAND_BFD)
    {
        item = cJSON_GetObjectItem(root2, "interval_ms");
        if (!item || !cJSON_IsNumber(item))
        {
            LOG_ERROR("%s: unexpected sf profile: (invalid health_check->interval_ms param) %s", LOG_TAG_POLICY, table_line);
            goto error_out;
        }
        param->sf_health_check.interval_ms = item->valueint;
        LOG_DEBUG("%s: parse sf profile: %d, health_check->interval_ms: %d", LOG_TAG_POLICY, param->sf_profile_id, item->valueint);

        item = cJSON_GetObjectItem(root2, "retires");
        if (!item || !cJSON_IsNumber(item))
        {
            LOG_ERROR("%s: unexpected sf profile: (invalid health_check->retires param) %s", LOG_TAG_POLICY, table_line);
            goto error_out;
        }
        param->sf_health_check.retires = item->valueint;
        LOG_DEBUG("%s: parse sf profile: %d, health_check->retires: %d", LOG_TAG_POLICY, param->sf_profile_id, item->valueint);
    }
    if (param->sf_connectivity.method != ENCAPSULATE_METHOD_LAYER2_SWITCH)
    {
        param->health_check_session_id = health_check_session_add(param->sf_profile_id, param->sf_vsys_id, &param->sf_health_check);
    }

    *ad = param;
    LOG_INFO("%s: Add sf profile: %d", LOG_TAG_POLICY, param->sf_profile_id);

    cJSON_Delete(root0);
    cJSON_Delete(root1);
    cJSON_Delete(root2);
    return;

error_out:
    if (root0)
    {
        cJSON_Delete(root0);
        root0 = NULL;
    }

    if (root1)
    {
        cJSON_Delete(root1);
        root1 = NULL;
    }

    if (root2)
    {
        cJSON_Delete(root2);
        root2 = NULL;
    }

    if (param)
    {
        free(param);
        param = NULL;
    }
}

static void sf_param_free_cb(int table_id, void **ad, long argl, void *argp)
{
    struct sf_param *param = (struct sf_param *)*ad;
    if (param == NULL)
    {
        return;
    }

    if ((__sync_sub_and_fetch(&param->sf_ref_cnt, 1) == 0))
    {
        if (param->sf_connectivity.method != ENCAPSULATE_METHOD_LAYER2_SWITCH)
        {
            health_check_session_del(param->health_check_session_id, param->sf_profile_id);
        }
        LOG_INFO("%s: Del sf profile: %d", LOG_TAG_POLICY, param->sf_profile_id);
        free(param);
        param = NULL;

        *ad = NULL;
    }
}

static void sf_param_dup_cb(int table_id, void **to, void **from, long argl, void *argp)
{
    struct sf_param *param = (struct sf_param *)*from;
    if (param)
    {
        __sync_add_and_fetch(&(param->sf_ref_cnt), 1);
        *to = param;
    }
    else
    {
        *to = NULL;
    }
}

static void sf_param_free(struct sf_param *param)
{
    sf_param_free_cb(0, (void **)&param, 0, NULL);
}

/******************************************************************************
 * Private API -- Selected SF
 ******************************************************************************/

static void selected_sf_init(struct selected_sf *selected_sf)
{
    if (selected_sf)
    {
        memset(selected_sf, 0, sizeof(struct selected_sf));
        selected_sf->rule_vsys_id = 0;
        selected_sf->rule_id = 0;
        selected_sf->traffic_type = TRAFFIC_TYPE_NONE;
        selected_sf->sff_profile_id = -1;
        selected_sf->sff_forward_type = FORWARD_TYPE_NONE;
        selected_sf->sf_profile_id = -1;
        selected_sf->sf_action = SESSION_ACTION_BYPASS;
        selected_sf->sf_action_desc = ACTION_BYPASS_DUE_DEFAULT;
    }
}

static void selected_sf_set_info(struct selected_sf *selected_sf, struct sf_param *sf_param)
{
    selected_sf->sf_vsys_id = sf_param->sf_vsys_id;
    selected_sf->sf_profile_id = sf_param->sf_profile_id;
    selected_sf->sf_connectivity = sf_param->sf_connectivity;

    if (selected_sf->sf_connectivity.method == ENCAPSULATE_METHOD_VXLAN_G)
    {
        selected_sf->sf_dst_ip = inet_addr(selected_sf->sf_connectivity.dest_ip);
    }
}

static void selected_sf_set_action(struct selected_sf *selected_sf, enum action_desc action_desc)
{
    selected_sf->sf_action_desc = action_desc;

    switch (action_desc)
    {
    case ACTION_BYPASS_DUE_DEFAULT:
    case ACTION_BYPASS_DUE_INVALID_POLICY:
    case ACTION_BYPASS_DUE_FAILURE_ACTION:
    case ACTION_BYPASS_DUE_UNAVAILABLE_ACTION:
    case ACTION_BYPASS_DUE_HEALTH_SF_LIMIT:
        selected_sf->sf_action = SESSION_ACTION_BYPASS;
        break;
    case ACTION_BLOCK_DUE_FAILURE_ACTION:
    case ACTION_BLOCK_DUE_UNAVAILABLE_ACTION:
        selected_sf->sf_action = SESSION_ACTION_BLOCK;
        break;
    case ACTION_FORWAED_DUE_SELECTED_SF:
        selected_sf->sf_action = SESSION_ACTION_FORWARD;
        break;
    }
}

// return  1 : current sf can   be selected
// return  0 : current sf can't be selected
static int select_sf_by_admin_status(struct sf_param *sf)
{
    if (sf->sf_admin_status == ADMMIN_STATUS_ACTIVE)
    {
        return 1;
    }
    else
    {
        return 0;
    }
}

// return  1 : current sf can   be selected
// return  0 : current sf can't be selected
static int select_sf_by_device_group(struct policy_enforcer *enforcer, struct sf_param *sf)
{
    if (strcasecmp(enforcer->config.device_group, sf->sf_effective_range.value) == 0)
    {
        return 1;
    }
    else
    {
        return 0;
    }
}

// return  1 : current sf can   be selected
// return  0 : current sf can't be selected
static int select_sf_by_data_center(struct policy_enforcer *enforcer, struct sf_param *sf)
{
    if (strcasecmp(enforcer->config.data_center, sf->sf_effective_range.value) == 0)
    {
        return 1;
    }
    else
    {
        return 0;
    }
}

// return  1 : current sf can   be selected
// return  0 : current sf can't be selected
static int select_sf_by_localization(struct policy_enforcer *enforcer, struct sff_param *sff_param, struct sf_param *sf)
{
    if (sff_param->sff_ldbc.localiza == LDBC_LOCALIZATION_NEARBY)
    {
        if (sf->sf_effective_range.type == EFFECTIVE_TYPE_DEVICE_GROUP)
        {
            return select_sf_by_device_group(enforcer, sf);
        }
        else
        {
            return select_sf_by_data_center(enforcer, sf);
        }
    }
    else
    {
        return 1;
    }
}

// return  1 : current sf can   be selected
// return  0 : current sf can't be selected
static int handle_fail_action(struct exception *sff_exception, struct selected_sf *selected_sf, int sf_num)
{
    if (sff_exception->fail_action == FAILURE_ACTION_RE_DISPATCH)
    {
        if (sff_exception->health_service_func_lt > 0 && sf_num < sff_exception->health_service_func_lt)
        {
            selected_sf_set_action(selected_sf, ACTION_BYPASS_DUE_HEALTH_SF_LIMIT);
            return 1;
        }
        else
        {
            if (sf_num == 0)
            {
                if (sff_exception->unavail_action == UNAVAILABLE_ACTION_BYPASSS)
                {
                    selected_sf_set_action(selected_sf, ACTION_BYPASS_DUE_UNAVAILABLE_ACTION);
                    return 1;
                }
                else
                {
                    selected_sf_set_action(selected_sf, ACTION_BLOCK_DUE_UNAVAILABLE_ACTION);
                    return 1;
                }
            }
            else
            {
                return 0;
            }
        }
    }
    else if (sff_exception->fail_action == FAILURE_ACTION_BYPASS)
    {
        selected_sf_set_action(selected_sf, ACTION_BYPASS_DUE_FAILURE_ACTION);
        return 1;
    }
    else if (sff_exception->fail_action == FAILURE_ACTION_BLOCK)
    {
        selected_sf_set_action(selected_sf, ACTION_BLOCK_DUE_FAILURE_ACTION);
        return 1;
    }
    else
    {
        return 0;
    }
}

static void select_sf_by_ldbc(struct sff_param *sff_param, struct selected_sf *selected_sf, struct session_ctx *s_ctx, UT_array *sf_array, uint64_t hash)
{
    struct thread_metrics *thread_metrics = &s_ctx->ref_thread_ctx->thread_metrics;

    while (utarray_len(sf_array))
    {
        unsigned int sf_index = (unsigned int)(hash % utarray_len(sf_array));
        struct sf_param *sf_param = (struct sf_param *)utarray_eltptr(sf_array, sf_index);

        if (sf_param->sf_connectivity.method == ENCAPSULATE_METHOD_LAYER2_SWITCH)
        {
            LOG_INFO("%s: session %lu %s select sf by ldbc, sf_profile_id %d to be selected", LOG_TAG_POLICY, s_ctx->session_id, s_ctx->session_addr, sf_param->sf_profile_id);
            selected_sf_set_action(selected_sf, ACTION_FORWAED_DUE_SELECTED_SF);
            selected_sf_set_info(selected_sf, sf_param);
            return;
        }

        memset(selected_sf->sf_dst_mac, 0, sizeof(selected_sf->sf_dst_mac));
        if (health_check_session_get_mac(sf_param->health_check_session_id, selected_sf->sf_dst_mac) == 0)
        {
            ATOMIC_INC(&(thread_metrics->sf_status.active));
            LOG_INFO("%s: session %lu %s select sf by ldbc, sf_profile_id %d to be selected", LOG_TAG_POLICY, s_ctx->session_id, s_ctx->session_addr, sf_param->sf_profile_id);
            selected_sf_set_action(selected_sf, ACTION_FORWAED_DUE_SELECTED_SF);
            selected_sf_set_info(selected_sf, sf_param);
            return;
        }
        else
        {
            ATOMIC_INC(&(thread_metrics->sf_status.inactive));
            if (handle_fail_action(&sff_param->sff_exception, selected_sf, utarray_len(sf_array) - 1) == 0)
            {
                LOG_INFO("%s: session %lu %s select sf by re-dispatch, sf_profile_id %d to be excluded", LOG_TAG_POLICY, s_ctx->session_id, s_ctx->session_addr, sf_param->sf_profile_id);
                utarray_erase(sf_array, sf_index, 1);
                continue;
            }
            else
            {
                LOG_INFO("%s: session %lu %s select sf by fail-action, sf_profile_id %d to be selected", LOG_TAG_POLICY, s_ctx->session_id, s_ctx->session_addr, sf_param->sf_profile_id);
                selected_sf_set_info(selected_sf, sf_param);
                return;
            }
        }
    }

    handle_fail_action(&sff_param->sff_exception, selected_sf, 0);
}

static void select_sf_from_sff(struct policy_enforcer *enforcer, struct sff_param *sff_param, struct selected_sf *selected_sf, struct session_ctx *s_ctx, uint64_t packet_hash)
{
    UT_array *sf_array;
    UT_icd sf_icd = {sizeof(struct sf_param), NULL, NULL, NULL};
    utarray_new(sf_array, &sf_icd);

    for (int i = 0; i < sff_param->sf_profile_ids_num; i++)
    {
        char profile_id[16] = {0};
        snprintf(profile_id, sizeof(profile_id), "%u", sff_param->sf_profile_ids[i]);

        struct sf_param *sf = (struct sf_param *)maat_plugin_table_get_ex_data(enforcer->maat, enforcer->sf_table_id, profile_id, strlen(profile_id));
        if (sf == NULL)
        {
            LOG_ERROR("%s: failed to get sf parameter of profile %d", LOG_TAG_POLICY, sff_param->sf_profile_ids[i]);
            continue;
        }

        if (select_sf_by_admin_status(sf) == 0)
        {
            LOG_INFO("%s: session %lu %s select sf by admin-status, sf_profile_id %d to be excluded", LOG_TAG_POLICY, s_ctx->session_id, s_ctx->session_addr, sf->sf_profile_id);
            sf_param_free(sf);
            continue;
        }

        if (select_sf_by_localization(enforcer, sff_param, sf) == 0)
        {
            LOG_INFO("%s: session %lu %s select sf by localization, sf_profile_id %d to be excluded", LOG_TAG_POLICY, s_ctx->session_id, s_ctx->session_addr, sf->sf_profile_id);
            sf_param_free(sf);
            continue;
        }

        utarray_push_back(sf_array, sf);
        sf_param_free(sf);
    }

    select_sf_by_ldbc(sff_param, selected_sf, s_ctx, sf_array, packet_hash);
    utarray_free(sf_array);
}

/******************************************************************************
 * Public API -- Utils
 ******************************************************************************/

const char *traffic_type_to_string(enum traffic_type traffic_type)
{
    switch (traffic_type)
    {
    case TRAFFIC_TYPE_NONE:
        return "none";
    case TRAFFIC_TYPE_RAW:
        return "raw";
    case TRAFFIC_TYPE_DECRYPTED:
        return "decrypted";
    default:
        return "unknown";
    }
}

const char *forward_type_to_string(enum forward_type forward_type)
{
    switch (forward_type)
    {
    case FORWARD_TYPE_NONE:
        return "none";
    case FORWARD_TYPE_STEERING:
        return "steering";
    case FORWARD_TYPE_MIRRORING:
        return "mirroring";
    default:
        return "unknown";
    }
}

const char *action_desc_to_string(enum action_desc action_desc)
{
    switch (action_desc)
    {
    case ACTION_BYPASS_DUE_DEFAULT:
        return "bypass(default)";
    case ACTION_BYPASS_DUE_HEALTH_SF_LIMIT:
        return "bypass(health_sf_limit)";
    case ACTION_BYPASS_DUE_UNAVAILABLE_ACTION:
        return "bypass(unavailable_action)";
    case ACTION_BYPASS_DUE_FAILURE_ACTION:
        return "bypass(failure_action)";
    case ACTION_BYPASS_DUE_INVALID_POLICY:
        return "bypass(invalid_policy)";
    case ACTION_BLOCK_DUE_UNAVAILABLE_ACTION:
        return "block(unavailable_action)";
    case ACTION_BLOCK_DUE_FAILURE_ACTION:
        return "block(failure_action)";
    case ACTION_FORWAED_DUE_SELECTED_SF:
        return "forward(selected_sf)";
    default:
        return "unknown";
    }
}

const char *encapsulate_method_to_string(enum encapsulate_method encap_method)
{
    switch (encap_method)
    {
    case ENCAPSULATE_METHOD_NONE:
        return "none";
    case ENCAPSULATE_METHOD_LAYER2_SWITCH:
        return "layer2_switch";
    case ENCAPSULATE_METHOD_LAYER3_SWITCH:
        return "layer3_switch";
    case ENCAPSULATE_METHOD_VXLAN_G:
        return "vxlan_g";
    default:
        return "unknown";
    }
}

/******************************************************************************
 * Public API -- Selected Chaining
 ******************************************************************************/

// return  NULL : error
// return !NULL : success
struct selected_chaining *selected_chaining_create(int chaining_size, uint64_t session_id, char *session_addr)
{
    struct selected_chaining *chaining = (struct selected_chaining *)calloc(1, sizeof(struct selected_chaining));
    assert(chaining);
    chaining->chaining_used = 0;
    chaining->chaining_size = chaining_size;
    chaining->chaining = (struct selected_sf *)calloc(chaining->chaining_size, sizeof(struct selected_sf));
    assert(chaining->chaining);

    for (int i = 0; i < chaining->chaining_size; i++)
    {
        struct selected_sf *item = &(chaining->chaining[i]);
        selected_sf_init(item);
    }
    chaining->session_id = session_id;
    chaining->session_addr = session_addr;

    return chaining;
}

void selected_chaining_destory(struct selected_chaining *chaining)
{
    if (chaining)
    {
        if (chaining->chaining)
        {
            free(chaining->chaining);
            chaining->chaining = NULL;
        }
        free(chaining);
        chaining = NULL;
    }
}

void selected_chaining_dump(struct selected_chaining *chaining)
{
    if (chaining == NULL)
    {
        LOG_DEBUG("%s: selected_chaining: NULL", LOG_TAG_POLICY);
        return;
    }

    LOG_DEBUG("%s: session %lu %s selected_chaining->chaining_size : %d", LOG_TAG_POLICY, chaining->session_id, chaining->session_addr, chaining->chaining_size);
    LOG_DEBUG("%s: session %lu %s selected_chaining->chaining_used : %d", LOG_TAG_POLICY, chaining->session_id, chaining->session_addr, chaining->chaining_used);

    for (int i = 0; i < chaining->chaining_used; i++)
    {
        struct selected_sf *node = &(chaining->chaining[i]);
        LOG_DEBUG("%s: session %lu %s selected_chaining->node[%d]->rule_id                             : %lu", LOG_TAG_POLICY, chaining->session_id, chaining->session_addr, i, node->rule_id);
        LOG_DEBUG("%s: session %lu %s selected_chaining->node[%d]->traffic_type                        : %s", LOG_TAG_POLICY, chaining->session_id, chaining->session_addr, i, traffic_type_to_string(node->traffic_type));
        // sff
        LOG_DEBUG("%s: session %lu %s selected_chaining->node[%d]->sff_profile_id                      : %d", LOG_TAG_POLICY, chaining->session_id, chaining->session_addr, i, node->sff_profile_id);
        LOG_DEBUG("%s: session %lu %s selected_chaining->node[%d]->sff_forward_type                    : %s", LOG_TAG_POLICY, chaining->session_id, chaining->session_addr, i, forward_type_to_string(node->sff_forward_type));
        // sf
        LOG_DEBUG("%s: session %lu %s selected_chaining->node[%d]->sf_profile_id                       : %d", LOG_TAG_POLICY, chaining->session_id, chaining->session_addr, i, node->sf_profile_id);
        LOG_DEBUG("%s: session %lu %s selected_chaining->node[%d]->sf_action_desc                      : %s", LOG_TAG_POLICY, chaining->session_id, chaining->session_addr, i, action_desc_to_string(node->sf_action_desc));
        LOG_DEBUG("%s: session %lu %s selected_chaining->node[%d]->sf_connectivity->encapsulate_method : %s", LOG_TAG_POLICY, chaining->session_id, chaining->session_addr, i, encapsulate_method_to_string(node->sf_connectivity.method));
        LOG_DEBUG("%s: session %lu %s selected_chaining->node[%d]->sf_connectivity->int_vlan_tag       : %d", LOG_TAG_POLICY, chaining->session_id, chaining->session_addr, i, node->sf_connectivity.int_vlan_tag);
        LOG_DEBUG("%s: session %lu %s selected_chaining->node[%d]->sf_connectivity->ext_vlan_tag       : %d", LOG_TAG_POLICY, chaining->session_id, chaining->session_addr, i, node->sf_connectivity.ext_vlan_tag);
        LOG_DEBUG("%s: session %lu %s selected_chaining->node[%d]->sf_connectivity->dest_ip            : %s", LOG_TAG_POLICY, chaining->session_id, chaining->session_addr, i, node->sf_connectivity.dest_ip);
    }
}

void selected_chaining_bref(struct selected_chaining *chaining)
{
    if (chaining == NULL)
    {
        return;
    }

    char buff[4096] = {0};
    int buff_used = 0;
    int buff_size = sizeof(buff);
    buff_used += snprintf(buff + buff_used, buff_size - buff_used, "chaining_size:%d, chaining_used:%d, {", chaining->chaining_size, chaining->chaining_used);
    for (int i = 0; i < chaining->chaining_used; i++)
    {
        struct selected_sf *node = &(chaining->chaining[i]);
        if (buff_size - buff_used > 0)
        {
            if (i != 0)
            {
                buff_used += snprintf(buff + buff_used, buff_size - buff_used, ",");
            }
            buff_used += snprintf(buff + buff_used, buff_size - buff_used,
                                  "\"node[%d]\":{\"policy\":\"%lu->%d->%d\",\"action\":\"%s->%s->%s\"}",
                                  i, node->rule_id, node->sff_profile_id, node->sf_profile_id,
                                  traffic_type_to_string(node->traffic_type), forward_type_to_string(node->sff_forward_type), action_desc_to_string(node->sf_action_desc));
        }
    }
    LOG_INFO("%s: session %lu %s selected_chaining_bref: %s}", LOG_TAG_POLICY, chaining->session_id, chaining->session_addr, buff);
}

void selected_chaining_uniq(struct selected_chaining *chaining)
{
    if (chaining == NULL)
    {
        return;
    }

    // Selected Service Chaining Before Unique : [1,2,3,1,2]
    // Selected Service Chaining After Unique : [1,2,3]

    int i = 0;
    int j = 0;
    int k = 0;
    int is_exist = 0;
    for (i = 0; i < chaining->chaining_used; i++)
    {
        is_exist = 0;
        for (j = 0; j < i; j++)
        {
            if (chaining->chaining[i].sf_profile_id == chaining->chaining[j].sf_profile_id && chaining->chaining[i].sf_action == chaining->chaining[j].sf_action)
            {
                is_exist = 1;
                break;
            }
        }
        if (is_exist == 0)
        {
            if (i != k)
            {
                memcpy(&(chaining->chaining[k]), &(chaining->chaining[i]), sizeof(struct selected_sf));
            }
            k++;
        }
    }

    chaining->chaining_used = k;
    // Selected Service Chaining After Unique : [1,2,3,1,2] -> [1,2,3,0,0]
    for (i = chaining->chaining_used; i < chaining->chaining_size; i++)
    {
        selected_sf_init(&(chaining->chaining[i]));
    }
}

/******************************************************************************
 * Public API -- Policy Enforcer
 ******************************************************************************/

// return  NULL : error
// return !NULL : success
struct policy_enforcer *policy_enforcer_create(const char *instance, const char *profile, int thread_num, void *logger)
{
    int ret = 0;
    int redis_port_begin = 0;
    int redis_port_end = 0;
    int redis_port_select = 0;

    struct policy_enforcer *enforcer = (struct policy_enforcer *)calloc(1, sizeof(struct policy_enforcer));
    assert(enforcer);
    policy_enforcer_config(profile, &(enforcer->config));

    struct maat_options *opts = maat_options_new();
    if (opts == NULL)
    {
        LOG_ERROR("%s: unable create maat opts", LOG_TAG_POLICY);
        goto error_out;
    }

    maat_options_set_logger(opts, "log/maat.log", (enum log_level)enforcer->config.log_level);
    maat_options_set_instance_name(opts, instance);
    maat_options_set_caller_thread_number(opts, thread_num);
    maat_options_set_foreign_cont_dir(opts, enforcer->config.foreign_cont_dir);
    maat_options_set_rule_effect_interval_ms(opts, enforcer->config.effect_interval_ms);
    // TODO set enforcer->config.scan_detail
    // Maat4 is not supported temporarily

    switch (enforcer->config.input_mode)
    {
    case MAAT_INPUT_JSON:
        if (!strlen(enforcer->config.json_cfg_file))
        {
            LOG_ERROR("%s: invalid json_cfg_file", LOG_TAG_POLICY);
            goto error_out;
        }
        maat_options_set_json_file(opts, enforcer->config.json_cfg_file);
        break;
    case MAAT_INPUT_REDIS:
        if (!strlen(enforcer->config.redis_server))
        {
            LOG_ERROR("%s: invalid redis_server", LOG_TAG_POLICY);
            goto error_out;
        }
        ret = sscanf(enforcer->config.redis_port_range, "%d-%d", &redis_port_begin, &redis_port_end);
        if (ret == 1)
        {
            redis_port_select = redis_port_begin;
        }
        else if (ret == 2)
        {
            srand(time(NULL));
            redis_port_select = redis_port_begin + rand() % (redis_port_end - redis_port_begin);
        }
        else
        {
            LOG_ERROR("%s: invalid redis_port_range", LOG_TAG_POLICY);
            goto error_out;
        }
        maat_options_set_redis(opts, enforcer->config.redis_server, redis_port_select, enforcer->config.redis_db_idx);
        break;
    case MAAT_INPUT_FILE:
        if (!strlen(enforcer->config.ful_cfg_dir))
        {
            LOG_ERROR("%s: invalid ful_cfg_dir", LOG_TAG_POLICY);
            goto error_out;
        }
        if (!strlen(enforcer->config.inc_cfg_dir))
        {
            LOG_ERROR("%s: invalid inc_cfg_dir", LOG_TAG_POLICY);
            goto error_out;
        }
        maat_options_set_iris(opts, enforcer->config.ful_cfg_dir, enforcer->config.inc_cfg_dir);
        break;
    default:
        LOG_ERROR("%s: invalid input_mode %d", LOG_TAG_POLICY, enforcer->config.input_mode);
        goto error_out;
    }

    if (enforcer->config.stat_switch)
    {
        maat_options_set_stat_on(opts);
        maat_options_set_stat_file(opts, enforcer->config.stat_file);
        if (enforcer->config.perf_switch)
        {
            maat_options_set_perf_on(opts);
        }
    }

    if (enforcer->config.deferred_load)
    {
        maat_options_set_deferred_load_on(opts);
    }

    if (strlen(enforcer->config.accept_tags))
    {
        maat_options_set_accept_tags(opts, enforcer->config.accept_tags);
    }

    enforcer->maat = maat_new(opts, enforcer->config.table_info);
    if (enforcer->maat == NULL)
    {
        LOG_ERROR("%s: unable create maat", LOG_TAG_POLICY);
        goto error_out;
    }

    maat_options_free(opts);
    opts = NULL;
    return enforcer;

error_out:
    if (opts)
    {
        maat_options_free(opts);
        opts = NULL;
    }
    policy_enforcer_destory(enforcer);
    return NULL;
}

void policy_enforcer_destory(struct policy_enforcer *enforcer)
{
    if (enforcer)
    {
        if (enforcer->maat)
        {
            maat_free(enforcer->maat);
            enforcer->maat = NULL;
        }

        free(enforcer);
        enforcer = NULL;
    }
}

// return  0 : success
// return -1 : error
int policy_enforcer_register(struct policy_enforcer *enforcer)
{
    LOG_INFO("%s: register policy callback ...", LOG_TAG_POLICY);
    enforcer->compile_table_id = maat_get_table_id(enforcer->maat, "SERVICE_CHAINING_COMPILE");
    if (enforcer->compile_table_id < 0)
    {
        LOG_ERROR("%s: register SERVICE_CHAINING_COMPILE table failed", LOG_TAG_POLICY);
        return -1;
    }

    enforcer->sff_table_id = maat_get_table_id(enforcer->maat, "SERVICE_FUNCTION_FORWARDER_PROFILE");
    if (enforcer->sff_table_id < 0)
    {
        LOG_ERROR("%s: register SERVICE_FUNCTION_FORWARDER_PROFILE table ailed", LOG_TAG_POLICY);
        return -1;
    }

    enforcer->sf_table_id = maat_get_table_id(enforcer->maat, "SERVICE_FUNCTION_PROFILE");
    if (enforcer->sf_table_id < 0)
    {
        LOG_ERROR("%s: register SERVICE_FUNCTION_PROFILE table failed", LOG_TAG_POLICY);
        return -1;
    }

    if (maat_plugin_table_ex_schema_register(enforcer->maat, "SERVICE_CHAINING_COMPILE",
                                             chaining_param_new_cb,
                                             chaining_param_free_cb,
                                             chaining_param_dup_cb,
                                             0, enforcer) != 0)
    {
        LOG_ERROR("%s: register SERVICE_CHAINING_COMPILE plugin extension callbacks failed", LOG_TAG_POLICY);
        return -1;
    }

    if (maat_plugin_table_ex_schema_register(enforcer->maat, "SERVICE_FUNCTION_FORWARDER_PROFILE",
                                             sff_param_new_cb,
                                             sff_param_free_cb,
                                             sff_param_dup_cb,
                                             0, enforcer) != 0)
    {
        LOG_ERROR("%s: register SERVICE_FUNCTION_FORWARDER_PROFILE plugin extension callbacks failed", LOG_TAG_POLICY);
        return -1;
    }

    if (maat_plugin_table_ex_schema_register(enforcer->maat, "SERVICE_FUNCTION_PROFILE",
                                             sf_param_new_cb,
                                             sf_param_free_cb,
                                             sf_param_dup_cb,
                                             0, enforcer) != 0)
    {
        LOG_ERROR("%s: register SERVICE_FUNCTION_PROFILE plugin extension callbacks failed", LOG_TAG_POLICY);
        return -1;
    }
    LOG_INFO("%s: register policy callback success", LOG_TAG_POLICY);

    return 0;
}

int policy_enforce_chaining_size(struct policy_enforcer *enforcer)
{
    return enforcer->config.max_chaining_size;
}

void policy_enforce_select_chainings(struct policy_enforcer *enforcer, struct selected_chainings *chainings, struct session_ctx *s_ctx, struct packet *data_pkt, uint64_t rule_id, int dir_is_i2e)
{
    char buffer[16] = {0};
    struct selected_chaining *chaining = NULL;

    snprintf(buffer, sizeof(buffer), "%lu", rule_id);
    struct chaining_param *chaining_param = (struct chaining_param *)maat_plugin_table_get_ex_data(enforcer->maat, enforcer->compile_table_id, buffer, strlen(buffer));
    if (chaining_param == NULL)
    {
        LOG_ERROR("%s: session %lu %s failed to get chaining parameter of policy %lu", LOG_TAG_POLICY, s_ctx->session_id, s_ctx->session_addr, rule_id);
        return;
    }

    if (chaining_param->traffic_type == TRAFFIC_TYPE_RAW)
    {
        chaining = chainings->chaining_raw;
    }
    else
    {
        chaining = chainings->chaining_decrypted;
    }
    LOG_INFO("%s: session %lu %s enforce %s chaining: rule_id %lu", LOG_TAG_POLICY, chaining->session_id, chaining->session_addr, traffic_type_to_string(chaining_param->traffic_type), rule_id);

    for (int i = 0; i < chaining_param->sff_profile_ids_num && chaining->chaining_used < chaining->chaining_size; i++)
    {
        struct selected_sf *selected_sf = &(chaining->chaining[chaining->chaining_used]);
        selected_sf_init(selected_sf);

        memset(buffer, 0, sizeof(buffer));
        snprintf(buffer, sizeof(buffer), "%u", chaining_param->sff_profile_ids[i]);
        struct sff_param *sff_param = (struct sff_param *)maat_plugin_table_get_ex_data(enforcer->maat, enforcer->sff_table_id, buffer, strlen(buffer));
        if (sff_param == NULL)
        {
            LOG_ERROR("%s: session %lu %s failed to get sff parameter of profile %d, bypass current sff !!!", LOG_TAG_POLICY, chaining->session_id, chaining->session_addr, chaining_param->sff_profile_ids[i]);
            continue;
        }

        // sc info
        selected_sf->rule_id = rule_id;
        selected_sf->rule_vsys_id = chaining_param->vsys_id;
        selected_sf->traffic_type = chaining_param->traffic_type;

        // sff info
        selected_sf->sff_profile_id = chaining_param->sff_profile_ids[i];
        selected_sf->sff_forward_type = sff_param->sff_forward_type;

        // sf_index
        selected_sf->sf_index = chaining->chaining_used;

        uint64_t packet_hash = packet_get_hash(data_pkt, sff_param->sff_ldbc.method, dir_is_i2e);
        select_sf_from_sff(enforcer, sff_param, selected_sf, s_ctx, packet_hash);

        LOG_INFO("%s: session %lu %s enforce chaining [%d/%d]: policy: %lu->%d->%d, action: %s->%s->%s",
                 LOG_TAG_POLICY, chaining->session_id, chaining->session_addr,
                 selected_sf->sf_index, chaining->chaining_size,
                 selected_sf->rule_id, selected_sf->sff_profile_id, selected_sf->sf_profile_id,
                 traffic_type_to_string(chaining_param->traffic_type), forward_type_to_string(selected_sf->sff_forward_type), action_desc_to_string(selected_sf->sf_action_desc));

        chaining->chaining_used++;
        sff_param_free(sff_param);
    }

    selected_chaining_uniq(chaining);
    chaining_param_free(chaining_param);
}