path: root/server/doris_server_scandir.cpp

#include <dirent.h> 
#include <stdlib.h>
#include <stdio.h>
#include <string.h>
#include <sys/stat.h>
#include <openssl/evp.h>
#include <openssl/md5.h>
#include <assert.h>
#include <unistd.h>

#include <MESA/MESA_handle_logger.h>

#include "doris_server_scandir.h"

#ifndef __FILENAME__
#define __FILENAME__ __FILE__
#endif
#define MESA_RUNTIME_LOGV4(handle, lv, fmt, args...)   \
    MESA_handle_runtime_log((handle), (lv), "DorisServer", "%s:%d, " fmt, __FILENAME__, __LINE__, ##args)

int scandir_md5_final_string(MD5_CTX *c, char *result, unsigned int size)
{
	unsigned char md5[17]={0};
	int i;

	if(MD5_Final(md5, c) != 1)
	{
		return -1;
	}
	if(size < 33)
		return -1;

	for(i=0; i<16; i++)
	{
		sprintf(result + i*2, "%02x", md5[i]);
	}
	result[32] = '\0';
	return 0;
}

//replacement of glibc scandir, to adapt dictator malloc wrap
#define	ENLARGE_STEP	1024
int my_scandir(const char *dir, struct dirent ***namelist,
              int(*filter)(const struct dirent *),
              int(*compar)(const void *, const void *))
{
	DIR * od;
	int n = 0;
	int DIR_ENT_SIZE=ENLARGE_STEP;
	struct dirent ** list = NULL;
	struct dirent * p;
	struct dirent entry,*result;

	if((dir == NULL) || (namelist == NULL) || (od = opendir(dir))==NULL)
		return -1;

	list = (struct dirent **)malloc(DIR_ENT_SIZE*sizeof(struct dirent *));
	while(0==readdir_r(od,&entry,&result))
	{
		if(result==NULL)
		{
			break;
		}
		if( filter && !filter(&entry))
			continue;

		p = (struct dirent *)malloc(sizeof(struct dirent));
		memcpy((void *)p,(void *)(&entry),sizeof(struct dirent));
		list[n] = p;

		n++;
		if(n >= DIR_ENT_SIZE)
		{
			DIR_ENT_SIZE+=ENLARGE_STEP;
			list=(struct dirent **)realloc((void*)list,DIR_ENT_SIZE*sizeof(struct dirent *));
		}
	} 
	closedir(od); 
	*namelist = list;

	if(compar)
	{
		qsort((void *)*namelist,n,sizeof(struct dirent *),compar); 
	}
	return n; 

}

int filter_fn(const struct dirent * ent)
{
	//   files in xfs are not DT_REG.
	//   if(ent->d_type != DT_REG)
	//     return 0;

	return (strncmp(ent->d_name,"full_config_index",strlen("full_config_index")) == 0||
		strncmp(ent->d_name,"inc_config_index",strlen("inc_config_index")) == 0);
}

void remove_configs_of_index_file(const char *filename)
{
	FILE* fp=NULL;
	int ret=0;
	char line[MAX_CONFIG_LINE];
	struct cfg_table_info idx;

	fp=fopen(filename, "r");
	while(!feof(fp))
	{
		memset(line, 0, sizeof(line));
		fgets(line, sizeof(line), fp);
		ret=sscanf(line,"%[^ \t]%*[ \t]%d%*[ \t]%s", idx.table_name, &idx.cfg_num, idx.cfg_path);
		if(ret != 3)
		{
			continue;
		}
		if(!access(idx.cfg_path, F_OK))
		{
			remove(idx.cfg_path);
		}
	}
	fclose(fp);
}

void remove_configs_version_smaller(const char *file_dir, int64_t version_higher, int recursively, void *logger)
{
	struct dirent **namelist;
	int64_t config_seq;
	int32_t i, n=0;
	char update_str[32], index_path[256];

	if(version_higher <= 1)
	{
		return;
	}

	n = my_scandir(file_dir, &namelist, filter_fn, (int (*)(const void*, const void*))alphasort);
	if(n <= 0)
	{
		return ;
	}
	for(i=0; i<n; i++)
	{
		if((strcmp(namelist[i]->d_name, ".") == 0) || (strcmp(namelist[i]->d_name, "..") == 0) || (namelist[i]->d_type==DT_DIR))
		{
			continue;
		}
		if(strlen(namelist[i]->d_name) > 42)
		{
			continue;
		}
		if(sscanf(namelist[i]->d_name,"%[a-zA-Z]_config_index.%ld", update_str, &config_seq) != 2)
		{
			continue;
		}
		if(version_higher <= config_seq)
		{
			continue;
		}
		snprintf(index_path, 256, "%s/%s", file_dir, namelist[i]->d_name);
		if(recursively)
		{
			remove_configs_of_index_file(index_path);
		}
		remove(index_path);
		MESA_RUNTIME_LOGV4(logger, RLOG_LV_INFO, "config file %s removed initiatively.", index_path);
	}
}

/*�����С�������򣬷���ֵ:����˶��ٸ�*/
u_int32_t get_full_topN_max_versions(const char *file_dir, int64_t *ver_array, int32_t maxsize)
{
	struct dirent **namelist;
	int64_t config_seq, tmpval;
	int32_t curnum=0, i, j, n=0;
	char update_str[32];

	n = my_scandir(file_dir, &namelist, filter_fn, (int (*)(const void*, const void*))alphasort);
	if(maxsize<=0 || n <= 0)
	{
		return 0;
	}
	for(i=0; i<maxsize; i++)
	{
		ver_array[i] = 0;
	}
	for(i=0; i<n; i++)
	{
		if((strcmp(namelist[i]->d_name, ".") == 0) || (strcmp(namelist[i]->d_name, "..") == 0) || (namelist[i]->d_type==DT_DIR))
		{
			continue;
		}
		if(strlen(namelist[i]->d_name) > 42)
		{
			continue;
		}
		if(sscanf(namelist[i]->d_name,"%[a-zA-Z]_config_index.%ld", update_str, &config_seq) != 2)
		{
			continue;
		}
		if(strncasecmp(update_str, "full", strlen(update_str)))
		{
			continue;
		}

		if(ver_array[0] < config_seq)
		{
			ver_array[0] = config_seq;
			for(j=1; j<maxsize && (ver_array[j-1] > ver_array[j]); j++)
			{
				tmpval = ver_array[j-1];
				ver_array[j-1] = ver_array[j];
				ver_array[j] = tmpval;
			}
			if(curnum < maxsize) curnum++;
		}
	}
	return curnum;
}

enum DORIS_UPDATE_TYPE get_new_idx_path(long long current_version, const char *file_dir, void *logger, struct index_path_array **idx_path, int *idx_num)
{
	struct dirent **namelist;
	char update_str[32]={0};
	long long latest_ful_version=0,latest_inc_version=0,config_seq=0;
	int n=0,i=0, full_file_idx=0,inc_idx_num=0;
	enum DORIS_UPDATE_TYPE update_type=CFG_UPDATE_TYPE_NONE;
	struct index_path_array *tmpidx_path_array;
	struct index_version_array *tmpidx_ver_array;

	n = my_scandir(file_dir, &namelist, filter_fn, (int (*)(const void*, const void*))alphasort);
	if(n < 0)
	{
		MESA_RUNTIME_LOGV4(logger,RLOG_LV_FATAL, "scan dir error");
		return CFG_UPDATE_TYPE_NONE;
	}

	tmpidx_ver_array = (struct index_version_array*)calloc(sizeof(struct index_version_array), n);
	inc_idx_num=0;
	for(i=0;i<n;i++)
	{
		if((strcmp(namelist[i]->d_name, ".") == 0) || (strcmp(namelist[i]->d_name, "..") == 0) || (namelist[i]->d_type==DT_DIR))
		{
			continue;
		}
		if(strlen(namelist[i]->d_name) > 42)
		{
			MESA_RUNTIME_LOGV4(logger,RLOG_LV_FATAL, "config file %s filename too long,should like full_config_index.00000000000000000001", namelist[i]->d_name);
			continue;
		}
		if(sscanf(namelist[i]->d_name,"%[a-zA-Z]_config_index.%lld",update_str,&config_seq) != 2)
		{
			MESA_RUNTIME_LOGV4(logger,RLOG_LV_FATAL, "config file %s filename error,should like full_config_index.00000000000000000001", namelist[i]->d_name);
			continue;
		}
		if(strncasecmp(update_str,"full",strlen(update_str))==0)
		{
			if(config_seq>latest_ful_version)
			{
				latest_ful_version=config_seq;
				full_file_idx=i;
			}
		}
		else if(strncasecmp(update_str,"inc",strlen(update_str))==0)
		{
			if(config_seq > current_version)
			{
				tmpidx_ver_array[inc_idx_num].file_inx = i;
				tmpidx_ver_array[inc_idx_num].version  = config_seq;
				inc_idx_num++;
				if(config_seq > latest_inc_version)
				{
					latest_inc_version = config_seq;
				}
			}
		}
		else
		{
			MESA_RUNTIME_LOGV4(logger,RLOG_LV_FATAL, "config file %s,not full or inc config", namelist[i]->d_name);
		}
	}

	//full update
	if(latest_ful_version > current_version)
	{
		tmpidx_path_array = (struct index_path_array *)malloc(sizeof(struct index_path_array));
		snprintf(tmpidx_path_array->path, MAX_CONFIG_FN_LEN, "%s/%s", file_dir, namelist[full_file_idx]->d_name);
		tmpidx_path_array->version = latest_ful_version;
		*idx_num = 1;
		update_type = CFG_UPDATE_TYPE_FULL;
	}
	//inc update,it's possible that do inc after full update in this function,but we'll process it at next loop.
	else if(latest_inc_version > current_version)
	{
		tmpidx_path_array = (struct index_path_array *)malloc(sizeof(struct index_path_array)*inc_idx_num);
		for(i=0;i<inc_idx_num;i++)
		{
			snprintf(tmpidx_path_array[i].path, MAX_CONFIG_FN_LEN, "%s/%s", file_dir, namelist[tmpidx_ver_array[i].file_inx]->d_name);
			tmpidx_path_array[i].version = tmpidx_ver_array[i].version;
		}
		*idx_num = inc_idx_num;
		update_type = CFG_UPDATE_TYPE_INC;
	}
	*idx_path = tmpidx_path_array;

	free(tmpidx_ver_array);
	for(i=0;i<n;i++)
	{
		free(namelist[i]);
	}
	free(namelist);
	return update_type;
}

int cm_read_cfg_index_file(const char* path, struct cfg_table_info* idx/*OUT*/, int size, void* logger)
{
	FILE* fp=NULL;
	int ret=0,i=0;
	char line[MAX_CONFIG_LINE];
	struct stat file_info;

	fp=fopen(path, "r");
	while(!feof(fp))
	{
		memset(line, 0, sizeof(line));
		fgets(line, sizeof(line), fp);
		ret=sscanf(line,"%[^ \t]%*[ \t]%d%*[ \t]%s%*[ \t]%s", idx[i].table_name, &(idx[i].cfg_num), idx[i].cfg_path, idx[i].user_region);
		if((ret!=3 && ret!=4))//jump over empty line
		{
			continue;
		}
		ret=stat(idx[i].cfg_path, &file_info);
		if(ret!=0 || file_info.st_size==0) //���ļ����·�
		{
			MESA_RUNTIME_LOGV4(logger,RLOG_LV_FATAL, "%s of %s not exisit", idx[i].cfg_path, path);
			fclose(fp);
			return -1;
		}
		idx[i].filesize = file_info.st_size;
		i++;
		if(i==size)
		{
			MESA_RUNTIME_LOGV4(logger,RLOG_LV_FATAL, "Too much lines in %s", path);
			break;
		}
	}
	fclose(fp);
	return i;
}

bool doris_read_table_file(struct doris_idxfile_scanner *scanner, struct cfg_table_info *table,
	struct doris_callbacks *doris_cbs, const char* dec_key, void* logger)
{
	FILE *fp;
	size_t readlen, remainlen, oncesize;
	MD5_CTX md5ctx;
	char md5buffer[64], *filename;
	struct tablemeta meta;

	if((fp = fopen(table->cfg_path, "r")) == NULL)
	{
		MESA_RUNTIME_LOGV4(logger,RLOG_LV_FATAL, "fopen table file %s failed: %s", table->cfg_path, strerror(errno));
		return false;
	}
	MD5_Init(&md5ctx);

	meta.tablename = table->table_name;
	filename = strrchr(table->cfg_path, '/');
	meta.filename  = (filename!=NULL)?(filename + 1):table->cfg_path;
	meta.userregion = (table->user_region[0] != '\0')?table->user_region:NULL;
	meta.size = table->filesize;
	meta.cfgnum = table->cfg_num;
	doris_cbs->cfgfile_start(NULL, &meta, table->cfg_path, doris_cbs->userdata);

	remainlen = table->filesize;
	while(remainlen > 0)
	{
		oncesize = (remainlen >= ONCE_BUF_SIZE)?ONCE_BUF_SIZE:remainlen;
		readlen = fread(scanner->oncebuf, 1, oncesize, fp);
		assert(readlen == oncesize);
		remainlen -= readlen;
		doris_cbs->cfgfile_update(NULL, scanner->oncebuf, readlen, doris_cbs->userdata);
		MD5_Update(&md5ctx, scanner->oncebuf, readlen);
	}
	scandir_md5_final_string(&md5ctx, md5buffer, 64);
	doris_cbs->cfgfile_finish(NULL, md5buffer, doris_cbs->userdata);
	fclose(fp);
	return true;
}

cJSON *doris_index_version_start(int64_t version, struct cfg_table_info *table_array, int table_num,
	enum DORIS_UPDATE_TYPE update_type, struct doris_callbacks *doris_cbs)
{
	cJSON *meta, *array, *tmp;

	meta = cJSON_CreateObject();
	cJSON_AddNumberToObject(meta, "type", update_type);
	cJSON_AddNumberToObject(meta, "version", version);

	array = cJSON_CreateArray();
	for(int i=0; i<table_num; i++)
	{
		tmp = cJSON_CreateObject();
		cJSON_AddStringToObject(tmp, "tablename", table_array[i].table_name);
		cJSON_AddNumberToObject(tmp, "size", table_array[i].filesize);
		cJSON_AddNumberToObject(tmp, "cfg_num", table_array[i].cfg_num);
		if(table_array->user_region[0] != '\0')
		{
			cJSON_AddStringToObject(tmp, "user_region", table_array[i].user_region);
		}
		cJSON_AddItemToArray(array, tmp);
	}
	cJSON_AddItemToObject(meta, "configs", array);

	doris_cbs->version_start(NULL, meta, doris_cbs->userdata);
	return meta;
}

enum DORIS_UPDATE_TYPE doris_index_file_traverse(struct doris_idxfile_scanner *scanner, const char*idx_dir,
	struct doris_callbacks *doris_cbs, const char* dec_key, void* logger)
{
	enum DORIS_UPDATE_TYPE update_type;
	struct index_path_array *idx_path_array=NULL;
	int idx_num=0,table_num=0,i=0,j=0;
	struct cfg_table_info table_array[CM_MAX_TABLE_NUM];
	bool update_rslt;

	memset(table_array,0,sizeof(table_array));
	update_type=get_new_idx_path(scanner->cur_version, idx_dir,logger, &idx_path_array, &idx_num);
	if(update_type == CFG_UPDATE_TYPE_NONE)
	{
		MESA_RUNTIME_LOGV4(logger,RLOG_LV_DEBUG, "Scan over, no new configs found, waiting next.", idx_path_array[i].path);
		return update_type;
	}

	update_rslt = true;
	for(i=0; i<idx_num; i++)
	{
		MESA_RUNTIME_LOGV4(logger, RLOG_LV_INFO, "load %s", idx_path_array[i].path);
		table_num=cm_read_cfg_index_file(idx_path_array[i].path, table_array, CM_MAX_TABLE_NUM, logger);
		if(table_num<=0)
		{
			MESA_RUNTIME_LOGV4(logger,RLOG_LV_FATAL, "\033[1;31;40m[Alert] Load %s failed, skip this wrong version!!!!\033[0m\n", idx_path_array[i].path);
			update_type = CFG_UPDATE_TYPE_ERR;
			scanner->cur_version = idx_path_array[i].version; //����İ汾����
			break;
		}
		scanner->cur_version = idx_path_array[i].version;

		cJSON *meta = doris_index_version_start(idx_path_array[i].version, table_array, table_num, update_type, doris_cbs);
		for(j=0; j<table_num && update_rslt; j++)
		{
			update_rslt = update_rslt && doris_read_table_file(scanner, table_array+j, doris_cbs, dec_key, logger);
		}
		if(update_rslt)
		{
			doris_cbs->version_finish(NULL, doris_cbs->userdata);
		}
		else
		{
			update_type = CFG_UPDATE_TYPE_ERR;
			doris_cbs->version_error(NULL, doris_cbs->userdata);
			cJSON_Delete(meta);
			MESA_RUNTIME_LOGV4(logger,RLOG_LV_FATAL, "\033[1;31;40m[Alert] Load %s failed, skip this wrong version!!!!\033[0m\n", idx_path_array[i].path);
			break;
		}
		cJSON_Delete(meta);
	}
	free(idx_path_array);
	return update_type;
}

struct doris_idxfile_scanner *doris_index_file_scanner(int64_t start_version)
{
	struct doris_idxfile_scanner *scanner;

	scanner = (struct doris_idxfile_scanner *)malloc(sizeof(struct doris_idxfile_scanner));
	scanner->cur_version = start_version;
	return scanner;
}