tango-tfe/common/src/tfe_fieldstat.cpp

#include <stdlib.h>
#include <tfe_fieldstat.h>

#include "tfe_stream.h"
#include "tfe_resource.h"

void tfe_set_intercept_metric(struct tfe_fieldstat_metric_t *fieldstat, struct tfe_cmsg *cmsg, int hit_count, int downstream_rx_pkts, int downstream_rx_bytes, int upstream_rx_pkts, int upstream_rx_bytes, int thread_id)
{
    int ret;
    uint16_t out_size;

    if (cmsg == NULL)
    {
        return;
    }

    int vsys_id = 0;
    ret = tfe_cmsg_get_value(cmsg, TFE_CMSG_POLICY_VSYS_ID, (unsigned char *)&vsys_id, sizeof(vsys_id), &out_size);
    if (ret != 0)
    {
        TFE_LOG_ERROR(g_default_logger, "failed at fetch vsys_id from cmsg: %s", strerror(-ret));
        return;
    }

    uint64_t rule_id = 0;
    ret = tfe_cmsg_get_value(cmsg, TFE_CMSG_POLICY_ID, (unsigned char *)&rule_id, sizeof(rule_id), &out_size);
    if (ret != 0)
    {
        TFE_LOG_ERROR(g_default_logger, "failed at fetch rule_id from cmsg: %s", strerror(-ret));
        return;
    }

    uint8_t hit_no_intercept = 0;
    ret = tfe_cmsg_get_value(cmsg, TFE_CMSG_HIT_NO_INTERCEPT, (unsigned char *)&hit_no_intercept, sizeof(hit_no_intercept), &out_size);
    if (ret != 0)
    {
        TFE_LOG_ERROR(g_default_logger, "failed at fetch hit_no_intercept from cmsg: %s", strerror(-ret));
        return;
    }

    // according to KNI -> MESA_dir_link_to_human()
    // 'E' or 'e': 表示发包方向是从Internal to External.
    // 'I' or 'i': 表示发包方向是从External to Internal.
    unsigned int route_dir;
    ret = tfe_cmsg_get_value(cmsg, TFE_CMSG_COMMON_DIRECTION, (unsigned char *)&route_dir, sizeof(route_dir), &out_size);
    if (ret != 0)
    {
        TFE_LOG_ERROR(g_default_logger, "failed at fetch route_dir from cmsg: %s", strerror(-ret));
        return;
    }

    int dir_is_e2i = 0;
    switch (route_dir)
    {
    case 'e':
    /* fall through */
    case 'E':
        dir_is_e2i = 0;
        break;
    case 'i':
    /* fall through */
    case 'I':
        dir_is_e2i = 1;
        break;
    default:
        TFE_LOG_ERROR(g_default_logger, "failed at fetch route dir from cmsg: invalid route dir %c", route_dir);
        return;
    }

    int in_pkts = 0;
    int in_bytes = 0;
    int out_pkts = 0;
    int out_bytes = 0;

    // incoming  : E2I 的流量
    // outcoming : I2E 的流量
    // first_ctr_packet_dir <==> client hello packet dir
    if (dir_is_e2i == 1)
    {
        in_pkts = downstream_rx_pkts;
        in_bytes = downstream_rx_bytes;

        out_pkts = upstream_rx_pkts;
        out_bytes = upstream_rx_bytes;
    }
    else
    {
        in_pkts = upstream_rx_pkts;
        in_bytes = upstream_rx_bytes;

        out_pkts = downstream_rx_pkts;
        out_bytes = downstream_rx_bytes;
    }

    int nr_tags = 0;
    struct fieldstat_tag temp_tags[TAG_MAX] = {0};

    temp_tags[nr_tags].key = "vsys_id";
    temp_tags[nr_tags].value_type = 0;
    temp_tags[nr_tags].value_int = vsys_id;
    nr_tags++;

    temp_tags[nr_tags].key = "rule_id";
    temp_tags[nr_tags].value_type = 0;
    temp_tags[nr_tags].value_int = rule_id;
    nr_tags++;

    uint8_t pinning_status = 0;
    if (tfe_cmsg_get_value(cmsg, TFE_CMSG_SSL_PINNING_STATE, (unsigned char *)&pinning_status, sizeof(pinning_status), &out_size) == 0)
    {
        temp_tags[nr_tags].key = "pinning_status";
        temp_tags[nr_tags].value_type = 0;
        temp_tags[nr_tags].value_int = pinning_status;
        nr_tags++;
    }

    // action : 2 Intercept; 3 No Intercept
    temp_tags[nr_tags].key = "action";
    temp_tags[nr_tags].value_type = 0;
    temp_tags[nr_tags].value_int = (hit_no_intercept == 1 ? 3 : 2);
    nr_tags++;

    // sub_action not need for intercept metrics

    if (hit_count > 0)
    {
        fieldstat_dynamic_table_metric_value_incrby(fieldstat->instance, fieldstat->table_id, fieldstat->column_array[COLUMN_HIT_COUNT], "proxy_rule_hits", hit_count, temp_tags, (size_t)nr_tags, thread_id);
    }

    if (in_pkts > 0)
    {
        fieldstat_dynamic_table_metric_value_incrby(fieldstat->instance, fieldstat->table_id, fieldstat->column_array[COLUMN_IN_PKTS], "proxy_rule_hits", in_pkts, temp_tags, (size_t)nr_tags, thread_id);
    }

    if (in_bytes > 0)
    {
        fieldstat_dynamic_table_metric_value_incrby(fieldstat->instance, fieldstat->table_id, fieldstat->column_array[COLUMN_IN_BYTES], "proxy_rule_hits", in_bytes, temp_tags, (size_t)nr_tags, thread_id);
    }

    if (out_pkts > 0)
    {
        fieldstat_dynamic_table_metric_value_incrby(fieldstat->instance, fieldstat->table_id, fieldstat->column_array[COLUMN_OUT_PKTS], "proxy_rule_hits", out_pkts, temp_tags, (size_t)nr_tags, thread_id);
    }

    if (out_bytes > 0)
    {
        fieldstat_dynamic_table_metric_value_incrby(fieldstat->instance, fieldstat->table_id, fieldstat->column_array[COLUMN_OUT_BYTES], "proxy_rule_hits", out_bytes, temp_tags, (size_t)nr_tags, thread_id);
    }
}

int tfe_fieldstat_metric_incrby(struct tfe_fieldstat_metric_t *fieldstat, unsigned int column_id, long long value, const struct fieldstat_tag tags[], int n_tags, int thread_id)
{
	return fieldstat_dynamic_table_metric_value_incrby(fieldstat->instance, fieldstat->table_id, column_id, "proxy_rule_hits", value, tags, (size_t)n_tags, thread_id);
}

struct tfe_fieldstat_metric_t *tfe_fieldstat_metric_create(char *telegraf_ip, int telegraf_port, char *app_name, int cycle, int max_thread, void *local_logger)
{
	int i=0;

	struct fieldstat_tag metric_tags[TAG_MAX - 1] = {{"vsys_id", 0, -1}, {"rule_id", 0, -1}, {"action", 0, -1}, {"sub_action", 2, -1}};
	const char *column_field[COLUMN_MAX] = {"hit_count", "in_bytes", "out_bytes", "in_pkts", "out_pkts"};
	enum field_type column_type[COLUMN_MAX] = {FIELD_TYPE_COUNTER, FIELD_TYPE_COUNTER, FIELD_TYPE_COUNTER, FIELD_TYPE_COUNTER, FIELD_TYPE_COUNTER};

	struct tfe_fieldstat_metric_t *fieldstat = ALLOC(struct tfe_fieldstat_metric_t, 1);

	fieldstat->instance = fieldstat_dynamic_instance_new(app_name, max_thread);
	if(!fieldstat->instance)
	{
		TFE_LOG_ERROR(local_logger, "fieldstat3 dynamic instance init failed.");
		return NULL;
	}

	fieldstat->max_thread=max_thread;
	fieldstat_dynamic_set_line_protocol_server(fieldstat->instance, telegraf_ip, telegraf_port);
	fieldstat_dynamic_set_output_interval(fieldstat->instance, cycle);

	fieldstat->table_id = fieldstat_register_dynamic_table(fieldstat->instance, "proxy_rule_hits", column_field, column_type, (size_t)COLUMN_MAX, fieldstat->column_array);
	if(fieldstat->table_id < 0)
	{
		TFE_LOG_ERROR(local_logger, "fieldstat3 register dynamic table failed.");
		FREE(&fieldstat);
		return NULL;
	}

	fieldstat->tags = ALLOC(struct fieldstat_tag*, max_thread);
	for (i = 0; i < max_thread; i++)
	{
		fieldstat->tags[i] = ALLOC(struct fieldstat_tag, TAG_MAX-1);
		memcpy(fieldstat->tags[i], metric_tags, sizeof(struct fieldstat_tag) * (size_t)(TAG_MAX-1));
	}

	fieldstat_dynamic_instance_start(fieldstat->instance);
	return fieldstat;
}

void tfe_fieldstat_metric_destroy(struct tfe_fieldstat_metric_t *fieldstat)
{
	if(fieldstat)
	{
		if(fieldstat->instance)
		{
			fieldstat_dynamic_instance_free(fieldstat->instance);
		}

		for (int i = 0; i < fieldstat->max_thread; i++)
		{
			if (fieldstat->tags[i])
			{
				FREE(&fieldstat->tags[i]);
			}
		}
		FREE(&fieldstat->tags);
		FREE(&fieldstat);
	}
}