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

#include <errno.h>
#include <assert.h>
#include <unistd.h>
#include <arpa/inet.h>
#include <sys/socket.h>
#include <MESA/MESA_prof_load.h>

#include "log.h"
#include "utils.h"
#include "sf_status.h"

#define SCE_SF_STATUS "SCE-SF-STATUS,sf_profile_id=%d,type=service_function_status sf_status=%d,sf_latency_us=%d"

struct node
{
    int sf_profile_id;
    int sf_status;
    int sf_latency;

    UT_hash_handle hh;
};

struct sf_status_config
{
    int enable;
    int interval_s;
    int telegraf_listen_port;
    char telegraf_bind_address[2048];
};

struct sf_status
{
    struct sf_status_config config;
    struct sockaddr_in sock_addr;
    int sockfd;

    struct node *htable;
    uint64_t htable_elem_count;
};

static void sf_status_parse_config(const char *profile, struct sf_status_config *config)
{
    MESA_load_profile_int_def(profile, "METRICS", "enable", &(config->enable), 1);
    MESA_load_profile_int_def(profile, "METRICS", "interval_s", &(config->interval_s), 1);
    MESA_load_profile_int_def(profile, "METRICS", "telegraf_listen_port", &(config->telegraf_listen_port), 8300);
    MESA_load_profile_string_def(profile, "METRICS", "telegraf_bind_address", config->telegraf_bind_address, sizeof(config->telegraf_bind_address), "127.0.0.1");

    LOG_DEBUG("%s: METRICS->enable                 : %d", LOG_TAG_SF_STATUS, config->enable);
    LOG_DEBUG("%s: METRICS->interval_s             : %d", LOG_TAG_SF_STATUS, config->interval_s);
    LOG_DEBUG("%s: METRICS->telegraf_listen_port   : %d", LOG_TAG_SF_STATUS, config->telegraf_listen_port);
    LOG_DEBUG("%s: METRICS->telegraf_bind_address  : %s", LOG_TAG_SF_STATUS, config->telegraf_bind_address);
}

void sf_status_destory(struct sf_status *handle)
{
    if (handle)
    {
        if (handle->sockfd)
        {
            close(handle->sockfd);
            handle->sockfd = -1;
        }

        struct node *temp = NULL;
        struct node *node = NULL;
        HASH_ITER(hh, handle->htable, node, temp)
        {
            HASH_DELETE(hh, handle->htable, node);
            free(node);
            node = NULL;
        }

        handle->htable_elem_count = 0;
        free(handle);
        handle = NULL;
    }
}

struct sf_status *sf_status_create(const char *profile)
{
    struct sf_status *handle = (struct sf_status *)calloc(1, sizeof(struct sf_status));
    assert(handle);
    sf_status_parse_config(profile, &(handle->config));

    if (handle->config.enable == 0)
    {
        return handle;
    }

    handle->sockfd = socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP);
    handle->sock_addr.sin_family = AF_INET;
    handle->sock_addr.sin_port = htons(handle->config.telegraf_listen_port);
    handle->sock_addr.sin_addr.s_addr = inet_addr(handle->config.telegraf_bind_address);
    handle->htable_elem_count = 0;
    if (handle->sockfd == -1)
    {
        LOG_ERROR("%s: failed to create udp sockfd %s:%d, errno: %d, %s", LOG_TAG_SF_STATUS, handle->config.telegraf_bind_address, handle->config.telegraf_listen_port, errno, strerror(errno));
        sf_status_destory(handle);
        return NULL;
    }

    return handle;
}

void sf_status_reset(struct sf_status *handle)
{
    if (handle->config.enable == 0)
    {
        return;
    }

    LOG_DEBUG("%s: reset: elem_num %lu", LOG_TAG_SF_STATUS, handle->htable_elem_count);

    struct node *temp = NULL;
    struct node *node = NULL;
    HASH_ITER(hh, handle->htable, node, temp)
    {
        HASH_DELETE(hh, handle->htable, node);

        free(node);
        node = NULL;
        handle->htable_elem_count--;
    }
}

void sf_status_delete(struct sf_status *handle, int sf_profile_id)
{
    if (handle->config.enable == 0)
    {
        return;
    }

    struct node *temp = NULL;
    HASH_FIND(hh, handle->htable, &sf_profile_id, sizeof(sf_profile_id), temp);
    if (temp)
    {
        handle->htable_elem_count--;
        LOG_DEBUG("%s: delete: sf_profile %d success, elem_num %lu", LOG_TAG_SF_STATUS, sf_profile_id, handle->htable_elem_count);
        HASH_DELETE(hh, handle->htable, temp);
    }
    else
    {
        LOG_DEBUG("%s: delete: sf_profile %d not exists, elem_num %lu", LOG_TAG_SF_STATUS, sf_profile_id, handle->htable_elem_count);
    }
}

void sf_status_update(struct sf_status *handle, int sf_profile_id, int sf_status, int sf_latency)
{
    if (handle->config.enable == 0)
    {
        return;
    }

    struct node *temp = NULL;
    HASH_FIND(hh, handle->htable, &sf_profile_id, sizeof(sf_profile_id), temp);
    if (temp)
    {
        LOG_DEBUG("%s: update: sf_profile %d status %d success, elem_num %lu", LOG_TAG_SF_STATUS, sf_profile_id, sf_status, handle->htable_elem_count);
        temp->sf_profile_id = sf_profile_id;
        temp->sf_status = sf_status;
        temp->sf_latency = sf_latency;
    }
    else
    {
        handle->htable_elem_count++;
        LOG_DEBUG("%s: insert: sf_profile %d status %d success, elem_num %lu", LOG_TAG_SF_STATUS, sf_profile_id, sf_status, handle->htable_elem_count);
        temp = (struct node *)calloc(1, sizeof(struct node));
        temp->sf_profile_id = sf_profile_id;
        temp->sf_status = sf_status;
        temp->sf_latency = sf_latency;

        HASH_ADD(hh, handle->htable, sf_profile_id, sizeof(sf_profile_id), temp);
    }
}

void sf_status_send(struct sf_status *handle)
{
    char buff[2048];
    int nsend = 0;
    int size = sizeof(buff);

    struct node *temp = NULL;
    struct node *node = NULL;

    if (handle->config.enable == 0)
    {
        return;
    }

    HASH_ITER(hh, handle->htable, node, temp)
    {
        memset(buff, 0, size);
        nsend = snprintf(buff, size, SCE_SF_STATUS,
                         node->sf_profile_id,
                         node->sf_status,
                         node->sf_latency);
        sendto(handle->sockfd, buff, nsend, 0, (struct sockaddr *)&handle->sock_addr, sizeof(handle->sock_addr));
    }
}

int sf_status_get_interval(struct sf_status *handle)
{
    return handle->config.interval_s;
}