pt_scaling.c

/*
 * Copyright (C) 2019 OpenSIPS Project
 *
 * This file is part of opensips, a free SIP server.
 *
 * opensips is free software; you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation; either version 2 of the License, or
 * (at your option) any later version
 *
 * opensips is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program; if not, write to the Free Software
 * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
 */

#include <sys/types.h>
#include <unistd.h>
#include <stdio.h>
#include "mem/shm_mem.h"
#include "socket_info.h"
#include "dprint.h"
#include "pt.h"
#include "ipc.h"
#include "daemonize.h"
#include "status_report.h"

struct process_group {
	enum process_type type;
	struct socket_info *si_filter;
	fork_new_process_f *fork_func;
	terminate_process_f *term_func;
	struct scaling_profile *prof;
	unsigned char history_size;
	unsigned char *history_map;
	unsigned char history_idx;
	unsigned short no_downscale_cycles;
	str sr_identifier_name;
	struct process_group *next;
};

static struct process_group *pg_head = NULL;

static struct scaling_profile *profiles_head = NULL;

static void *pts_sr_group = NULL;


int init_auto_scaling(void)
{
	if (!auto_scaling_enabled)
		return 0;

	pts_sr_group = sr_register_group(CHAR_INT("auto-scaling"), 0/*is_public*/);
	if (pts_sr_group==NULL) {
		LM_ERR("Failed to register 'status_report' group for "
			"'auto-scaling' support\n");
		return -1;
	}

	return 0;
}


int create_auto_scaling_profile( char *name,
	unsigned int max_procs, unsigned int up_threshold,
	unsigned int up_cycles_needed, unsigned int up_cycles_tocheck,
	unsigned int min_procs, unsigned int down_threshold,
	unsigned int down_cycles_tocheck, unsigned short down_cycles_delay)
{
	struct scaling_profile *p;

	/* check for duplicates */
	if ( get_scaling_profile(name) ) {
		LM_ERR("profile <%s> (case insensitive) already created"
			" - double definition?? \n", name);
		return -1;
	}

	/* some sanity checks */
	if (min_procs==0) {
		down_threshold = 0;
		down_cycles_tocheck = 0;
		down_cycles_delay = 0;
	}
	if (max_procs==0 || max_procs <= min_procs || max_procs>=1000) {
		LM_ERR("invalid relation or range for MIN/MAX processes [%d,%d]\n",
			min_procs, max_procs);
		return -1;
	}
	if (up_threshold==0 || up_threshold <= down_threshold ||
	up_threshold>100 || down_threshold>100) {
		LM_ERR("invalid relation or range DOWN/UP thresholds percentages "
			"[%d,%d]\n", down_threshold, up_threshold);
		return -1;
	}
	if (up_cycles_needed==0 || up_cycles_tocheck==0 ||
	up_cycles_tocheck<up_cycles_needed) {
		LM_ERR("invalid relation or values for upscaling check [%d of %d]\n",
			up_cycles_needed, up_cycles_tocheck);
		return -1;
	}

	/* all good, create it*/

	p = (struct scaling_profile*)pkg_malloc( sizeof(struct scaling_profile) +
		strlen(name) + 1 );
	if (p==NULL) {
		LM_ERR("failed to allocate memory for a new auto-scaling profile\n");
		return -1;
	}

	/* not really need, more to be safe for future expansions */
	memset( p, 0, sizeof(struct scaling_profile));

	p->max_procs = max_procs;
	p->up_threshold = up_threshold;
	p->up_cycles_needed = up_cycles_needed;
	p->up_cycles_tocheck = up_cycles_tocheck;
	p->min_procs = min_procs;
	p->down_threshold = down_threshold;
	p->down_cycles_tocheck = down_cycles_tocheck;
	p->down_cycles_delay = down_cycles_delay;
	p->name = (char*)(p+1);
	strcpy( p->name, name);

	LM_DBG("profile <%s> created UP [max=%d, th=%d%%, check %d/%d] DOWN "
		"[min=%d, th=%d%%, check %d, delay=%d]\n", name,
		max_procs, up_threshold, up_cycles_needed, up_cycles_tocheck,
		min_procs, down_threshold, down_cycles_tocheck, down_cycles_delay);

	p->next = profiles_head;
	profiles_head = p;

	return 0;
}


struct scaling_profile *get_scaling_profile(char *name)
{
	struct scaling_profile *p;

	for ( p=profiles_head ; p ; p=p->next )
		if (strcasecmp(name, p->name)==0)
			return p;

	return NULL;
}


#define get_sr_type_name(_type, _s) \
	do { \
		if (_type==TYPE_UDP) {              \
			(_s).s = "UDP"; (_s).len = 3;   \
		} else if (_type==TYPE_TCP) {       \
			(_s).s = "TCP"; (_s).len = 3;   \
		} else if (_type==TYPE_TIMER) {     \
			(_s).s = "TIMER"; (_s).len = 5; \
		} else {                            \
			LM_BUG("unsupported auto-scaling group %d\n", _type); \
			(_s).s = "??"; (_s).len = 2;    \
		}                                   \
	}while(0)


int create_process_group(enum process_type type,
		struct socket_info *si_filter, struct scaling_profile *prof,
		fork_new_process_f *f1, terminate_process_f *f2)
{
	struct process_group *pg, *it;
	int h_size;
	str type_s;

	/* how much of a history do we need in order to cover both up and down
	 * tranzitions ? */
	h_size = (prof->up_cycles_tocheck > prof->down_cycles_tocheck) ?
		prof->up_cycles_tocheck : prof->down_cycles_tocheck;

	get_sr_type_name( type, type_s);

	pg = (struct process_group*)shm_malloc( sizeof(struct process_group) +
		sizeof(char)*h_size +
		type_s.len+(si_filter?1+(si_filter->sock_str.len):0)
		);
	if (pg==NULL) {
		LM_ERR("failed to allocate memory for a new process group\n");
		return -1;
	}
	memset( pg, 0, sizeof(struct process_group) + sizeof(char)*h_size );

	pg->type = type;
	pg->si_filter = si_filter;
	pg->prof = prof;
	pg->fork_func = f1;
	pg->term_func = f2;
	pg->next = NULL;

	pg->history_size = h_size;
	pg->history_map = (unsigned char*)(pg+1);
	pg->history_idx = 0;
	pg->no_downscale_cycles = pg->prof->down_cycles_delay;

	pg->sr_identifier_name.s = (char*)(pg->history_map + h_size);
	memcpy( pg->sr_identifier_name.s, type_s.s, type_s.len);
	pg->sr_identifier_name.len = type_s.len;
	if (si_filter) {
		pg->sr_identifier_name.s[pg->sr_identifier_name.len++] = '/';
		memcpy( pg->sr_identifier_name.s + pg->sr_identifier_name.len,
			si_filter->sock_str.s, si_filter->sock_str.len);
		pg->sr_identifier_name.len += si_filter->sock_str.len;
	}

	LM_DBG("registering group of processes type %d, socket filter %p, "
		"name <%.*s> scaling profile <%s>\n", type, si_filter,
		pg->sr_identifier_name.len, pg->sr_identifier_name.s, prof->name);

	if (sr_register_identifier( pts_sr_group,
		pg->sr_identifier_name.s, pg->sr_identifier_name.len,
		SR_STATUS_READY, CHAR_INT("active"), 100)<0
	) {
		LM_ERR("failed to register auto-scaling identifier for group"
			"name <%.*s>, disabling\n",
		pg->sr_identifier_name.len, pg->sr_identifier_name.s);
		pg->sr_identifier_name.s = NULL;
		pg->sr_identifier_name.len = 0;
	}

	/* add at the end of list, to avoid changing the head of the list due
	 * forking */
	for( it=pg_head ; it && it->next ; it=it->next);
	if (it==NULL)
		pg_head = pg;
	else
		it->next = pg;

	return 0;
}


static void _pt_raise_event(struct process_group *pg, int p_id, int load,
																char *scale)
{
	static str pt_ev_type = str_init("group_type");
	static str pt_ev_filter = str_init("group_filter");
	static str pt_ev_load = str_init("group_load");
	static str pt_ev_scale = str_init("scale");
	static str pt_ev_p_id = str_init("process_id");
	static str pt_ev_pid = str_init("pid");
	evi_params_p list = NULL;
	str s;

	if (!evi_probe_event(EVI_PROC_AUTO_SCALE_ID))
		return;

	list = evi_get_params();
	if (!list) {
		LM_ERR("cannot create event params\n");
		return;
	}

	get_sr_type_name( pg->type, s);

	if (evi_param_add_str(list, &pt_ev_type, &s) < 0) {
		LM_ERR("cannot add group type\n");
		goto error;
	}

	if (pg->si_filter==NULL) {
		s.s = "none"; s.len = 4;
	} else {
		s = pg->si_filter->sock_str;
	}
	if (evi_param_add_str(list, &pt_ev_filter, &s) < 0) {
		LM_ERR("cannot add group filter\n");
		goto error;
	}

	if (evi_param_add_int(list, &pt_ev_load, &load) < 0) {
		LM_ERR("cannot add group load\n");
		goto error;
	}

	s.s = scale; s.len = strlen(s.s);
	if (evi_param_add_str(list, &pt_ev_scale, &s) < 0) {
		LM_ERR("cannot add scaling type\n");
		goto error;
	}

	if (evi_param_add_int(list, &pt_ev_p_id, &p_id) < 0) {
		LM_ERR("cannot add process id\n");
		goto error;
	}

	if (evi_param_add_int(list, &pt_ev_pid, &(pt[p_id].pid)) < 0) {
		LM_ERR("cannot add process pid\n");
		goto error;
	}

	if (evi_dispatch_event(EVI_PROC_AUTO_SCALE_ID, list)) {
		LM_ERR("unable to send auto scaling event\n");
	}
	return;

error:
	evi_free_params(list);
}


static void rescale_group_history(struct process_group *pg, unsigned int idx,
		int org_size, int offset)
{
	unsigned int k;
	unsigned char old;

	k = idx;
	do {
		old = pg->history_map[k] ;
		pg->history_map[k] = (pg->history_map[k]*org_size)/(org_size+offset);
		LM_DBG("rescaling old %d to %d [idx %d]\n",
			old, pg->history_map[k], k);

		k = k ? (k-1) : (pg->history_size-1) ;
	} while(k!=idx);
}


void do_workers_auto_scaling(void)
{
	struct process_group *pg;
	unsigned int i, k, idx;
	unsigned int load;
	unsigned int procs_no;
	unsigned char cnt_under, cnt_over;
	int p_id, last_idx_in_pg;

	/* iterate all the groups we have */
	for ( pg=pg_head ; pg ; pg=pg->next ) {

		load = 0;
		procs_no = 0;
		last_idx_in_pg = -1;

		/* find the processes belonging to this group */
		for ( i=0 ; i<counted_max_processes ; i++) {

			/* skip processes:
			 * - not running
			 * - runing, but marked for termination
			 * - not part of the group
			 * - with a different group filter (socket interface) */
			if (!is_process_running(i) || pt[i].flags&OSS_PROC_TO_TERMINATE ||
			pt[i].type != pg->type || pg->si_filter!=pt[i].pg_filter)
				continue;

			load += get_stat_val( pt[i].load_rt );
			last_idx_in_pg = i;
			procs_no++;

		}

		if (!procs_no) {
			LM_BUG("no process beloging to group %d\n", pg->type);
			continue;
		}

		/* set the current value */
		idx = (pg->history_idx+1)%pg->history_size;
		pg->history_map[idx] = (unsigned char) ( load / procs_no );

		//LM_DBG("group %d (with %d procs) has average load of %d\n",
		//	pg->type, procs_no, pg->history_map[idx]);

		/* do the check over the history */
		cnt_over = 0;
		cnt_under = 0;
		k = idx;
		i = 1;
		do {
			if ( pg->history_map[k] > pg->prof->up_threshold &&
			i <= pg->prof->up_cycles_tocheck )
				cnt_over++;
			else if ( pg->history_map[k] < pg->prof->down_threshold &&
			i <= pg->prof->down_cycles_tocheck )
				cnt_under++;

			i++;
			k = k ? (k-1) : (pg->history_size-1) ;
		} while(k!=idx);

		/* decide what to do */
		if ( cnt_over >= pg->prof->up_cycles_needed ) {
			if ( procs_no < pg->prof->max_procs ) {
				LM_NOTICE("score %d/%d -> forking new proc in group %d "
					"(with %d procs)\n", cnt_over, pg->prof->up_cycles_tocheck,
					pg->type, procs_no);
				/* we need to fork one more process here */
				if ( (p_id=pg->fork_func(pg->si_filter))<0 ||
				wait_for_one_child()<0 ) {
					LM_ERR("failed to fork new process for group %d "
						"(current %d procs)\n",pg->type,procs_no);
				} else {
					sr_add_report_fmt( pts_sr_group,
						pg->sr_identifier_name.s, pg->sr_identifier_name.len,
						0, "Forking new process id %d at load %d "
						"in group with %d processes",
						p_id, pg->history_map[idx], procs_no);
					_pt_raise_event( pg, p_id, pg->history_map[idx] ,"up");
					rescale_group_history( pg, idx, procs_no, +1);
					pg->no_downscale_cycles = pg->prof->down_cycles_delay;
				}
			}
		} else if ( pg->prof->down_cycles_tocheck != 0 &&
		cnt_under == pg->prof->down_cycles_tocheck ) {
			if ( procs_no > pg->prof->min_procs &&
			pg->no_downscale_cycles==0) {
				/* try to estimate the load after downscaling */
				load = 0;
				k = idx;
				i = 0;
				do {
					load += pg->history_map[k];
					k = k ? (k-1) : (pg->history_size-1) ;
					i++;
				} while( k != idx && i <= pg->prof->down_cycles_tocheck );
				load = (load*procs_no) /
					(pg->prof->down_cycles_tocheck * (procs_no-1));
				if ( load < pg->prof->up_threshold ) {
					/* down scale one more process here */
					LM_NOTICE("score %d/%d -> ripping proc %d from group %d "
						"(with %d procs), estimated load -> %d\n", cnt_under,
						pg->prof->down_cycles_tocheck, last_idx_in_pg,
						pg->type, procs_no, load );
					pt[last_idx_in_pg].flags |= OSS_PROC_TO_TERMINATE;
					ipc_send_rpc( last_idx_in_pg, pg->term_func, NULL);
					sr_add_report_fmt( pts_sr_group,
						pg->sr_identifier_name.s, pg->sr_identifier_name.len,
						0, "Ripping process id %d at load %d "
						"in group with %d processes",
						last_idx_in_pg, pg->history_map[idx], procs_no);
					_pt_raise_event( pg, last_idx_in_pg,
						pg->history_map[idx], "down");
				}
			}
		}

		pg->history_idx++;
		if (pg->no_downscale_cycles) pg->no_downscale_cycles--;
	}
}