|
|
@@ -76,12 +76,6 @@ class StatisticWorker extends Man\Core\SocketWorker
|
|
|
*/
|
|
|
public function dealProcess($recv_str)
|
|
|
{
|
|
|
- // 如果是JSON协议，则是请求统计数据
|
|
|
- if($recv_str[0] === '{')
|
|
|
- {
|
|
|
- return $this->dealProvider($recv_str);
|
|
|
- }
|
|
|
-
|
|
|
// 解码
|
|
|
$unpack_data = StatisticProtocol::decode($recv_str);
|
|
|
$module = $unpack_data['module'];
|
|
|
@@ -227,12 +221,6 @@ class StatisticWorker extends Man\Core\SocketWorker
|
|
|
\Man\Core\Lib\Task::add(self::CLEAR_PERIOD_LENGTH, array($this, 'clearDisk'), array(WORKERMAN_LOG_DIR . $this->statisticDir, self::EXPIRED_TIME));
|
|
|
\Man\Core\Lib\Task::add(self::CLEAR_PERIOD_LENGTH, array($this, 'clearDisk'), array(WORKERMAN_LOG_DIR . $this->logDir, self::EXPIRED_TIME));
|
|
|
|
|
|
- // 创建一个tcp监听，用来提供统计查询服务
|
|
|
- $this->providerSocket = stream_socket_server(\Man\Core\Lib\Config::get($this->workerName.'.provider_listen'));
|
|
|
- if($this->providerSocket)
|
|
|
- {
|
|
|
- $ret = $this->event->add($this->providerSocket, \Man\Core\Events\BaseEvent::EV_READ, array($this, 'accept'));
|
|
|
- }
|
|
|
}
|
|
|
|
|
|
/**
|
|
|
@@ -273,344 +261,6 @@ class StatisticWorker extends Man\Core\SocketWorker
|
|
|
}
|
|
|
}
|
|
|
|
|
|
- /**
|
|
|
- * 处理请求统计
|
|
|
- * @param string $recv_str
|
|
|
- */
|
|
|
- protected function dealProvider($recv_str)
|
|
|
- {
|
|
|
- $req_data = json_decode(trim($recv_str), true);
|
|
|
- $module = $req_data['module'];
|
|
|
- $interface = $req_data['interface'];
|
|
|
- $cmd = $req_data['cmd'];
|
|
|
- $start_time = isset($req_data['start_time']) ? $req_data['start_time'] : '';
|
|
|
- $end_time = isset($req_data['end_time']) ? $req_data['end_time'] : '';
|
|
|
- $date = isset($req_data['date']) ? $req_data['date'] : '';
|
|
|
- $code = isset($req_data['code']) ? $req_data['code'] : '';
|
|
|
- $msg = isset($req_data['msg']) ? $req_data['msg'] : '';
|
|
|
- $offset = isset($req_data['offset']) ? $req_data['offset'] : '';
|
|
|
- $count = isset($req_data['count']) ? $req_data['count'] : 10;
|
|
|
- switch($cmd)
|
|
|
- {
|
|
|
- case 'get_statistic':
|
|
|
- $buffer = json_encode(array('modules'=>$this->getModules($module), 'statistic' => $this->getStatistic($date, $module, $interface)))."\n";
|
|
|
- $this->tcpSendToClient($buffer);
|
|
|
- break;
|
|
|
- case 'get_log':
|
|
|
- $buffer = json_encode($this->getStasticLog($module, $interface , $start_time , $end_time, $code = '', $msg = '', $offset='', $count=10))."\n";
|
|
|
- $this->tcpSendToClient($buffer);
|
|
|
- break;
|
|
|
- default :
|
|
|
- $this->tcpSendToClient('pack err');
|
|
|
- }
|
|
|
- $fd = $this->currentDealFd;
|
|
|
- $this->event->del($this->connections[$fd], Man\Core\Events\BaseEvent::EV_READ);
|
|
|
- $this->event->del($this->connections[$fd], Man\Core\Events\BaseEvent::EV_WRITE);
|
|
|
- fclose($this->connections[$fd]);
|
|
|
- unset($this->connections[$fd], $this->recvBuffers[$fd], $this->sendBuffers[$fd]);
|
|
|
- }
|
|
|
-
|
|
|
- /**
|
|
|
- * 获取模块
|
|
|
- * @return array
|
|
|
- */
|
|
|
- public function getModules($current_module = '')
|
|
|
- {
|
|
|
- $st_dir = WORKERMAN_ROOT_DIR . $this->statisticDir;
|
|
|
- $modules_name_array = array();
|
|
|
- foreach(glob($st_dir."/*", GLOB_ONLYDIR) as $module_file)
|
|
|
- {
|
|
|
- $tmp = explode("/", $module_file);
|
|
|
- $module = end($tmp);
|
|
|
- $modules_name_array[$module] = array();
|
|
|
- if($current_module == $module)
|
|
|
- {
|
|
|
- $st_dir = $st_dir.$current_module.'/';
|
|
|
- $all_interface = array();
|
|
|
- foreach(glob($st_dir."*") as $file)
|
|
|
- {
|
|
|
- if(is_dir($file))
|
|
|
- {
|
|
|
- continue;
|
|
|
- }
|
|
|
- list($interface, $date) = explode("|", basename($file));
|
|
|
- $all_interface[$interface] = $interface;
|
|
|
- }
|
|
|
- $modules_name_array[$module] = $all_interface;
|
|
|
- }
|
|
|
- }
|
|
|
- return $modules_name_array;
|
|
|
- }
|
|
|
-
|
|
|
- /**
|
|
|
- * 获得统计数据
|
|
|
- * @param string $module
|
|
|
- * @param string $interface
|
|
|
- * @param int $date
|
|
|
- * @return bool/string
|
|
|
- */
|
|
|
- protected function getStatistic($date, $module, $interface)
|
|
|
- {
|
|
|
- if(empty($module) || empty($interface))
|
|
|
- {
|
|
|
- return '';
|
|
|
- }
|
|
|
- // log文件
|
|
|
- $log_file = WORKERMAN_LOG_DIR . $this->statisticDir."{$module}/{$interface}|{$date}";
|
|
|
- return @file_get_contents($log_file);
|
|
|
- }
|
|
|
-
|
|
|
-
|
|
|
- /**
|
|
|
- * 批量请求
|
|
|
- * @param array $request_buffer_array ['ip:port'=>req_buf, 'ip:port'=>req_buf, ...]
|
|
|
- * @return array
|
|
|
- */
|
|
|
- public function multiRequest($request_buffer_array)
|
|
|
- {
|
|
|
- $client_array = $sock_to_ip = $ip_list = array();
|
|
|
- foreach($request_buffer_array as $address => $buffer)
|
|
|
- {
|
|
|
- $client = stream_socket_client($address, $errno, $errmsg, 1);
|
|
|
- if(!$client)
|
|
|
- {
|
|
|
- $this->notice("connect $address fail");
|
|
|
- continue;
|
|
|
- }
|
|
|
- $client_array[$address] = $client;
|
|
|
- stream_set_timeout($client_array[$address], 0, 100000);
|
|
|
- fwrite($client_array[$address], $buffer);
|
|
|
- stream_set_blocking($client_array[$address], 0);
|
|
|
- $sock_to_address[(int)$client] = $address;
|
|
|
- }
|
|
|
- $read = $client_array;
|
|
|
- $write = $except = $read_buffer = array();
|
|
|
- $time_start = microtime(true);
|
|
|
- // 超时设置
|
|
|
- $timeout = 1;
|
|
|
- // 轮询处理数据
|
|
|
- while(count($read) > 0)
|
|
|
- {
|
|
|
- if(stream_select($read, $write, $except, $timeout))
|
|
|
- {
|
|
|
- foreach($read as $socket)
|
|
|
- {
|
|
|
- $address = $sock_to_address[(int)$socket];
|
|
|
- $buf = fread($socket, 8192);
|
|
|
- if(!$buf)
|
|
|
- {
|
|
|
- if(feof($socket))
|
|
|
- {
|
|
|
- unset($client_array[$address]);
|
|
|
- }
|
|
|
- continue;
|
|
|
- }
|
|
|
- if(!isset($read_buffer[$address]))
|
|
|
- {
|
|
|
- $read_buffer[$address] = $buf;
|
|
|
- }
|
|
|
- else
|
|
|
- {
|
|
|
- $read_buffer[$address] .= $buf;
|
|
|
- }
|
|
|
- // 数据接收完毕
|
|
|
- if("\n" === $read_buffer[$address][strlen($read_buffer[$address])-1])
|
|
|
- {
|
|
|
- unset($client_array[$address]);
|
|
|
- }
|
|
|
- }
|
|
|
- }
|
|
|
- // 超时了
|
|
|
- if(microtime(true) - $time_start > $timeout)
|
|
|
- {
|
|
|
- break;
|
|
|
- }
|
|
|
- $read = $client_array;
|
|
|
- }
|
|
|
- ksort($read_buffer);
|
|
|
- return $read_buffer;
|
|
|
- }
|
|
|
-
|
|
|
- /**
|
|
|
- * 获取指定日志
|
|
|
- *
|
|
|
- */
|
|
|
- protected function getStasticLog($module, $interface , $start_time = '', $end_time = '', $code = '', $msg = '', $offset='', $count=100)
|
|
|
- {
|
|
|
- // log文件
|
|
|
- $log_file = WORKERMAN_ROOT_DIR . $this->logDir. (empty($start_time) ? date('Y-m-d') : date('Y-m-d', $start_time));
|
|
|
- if(!is_readable($log_file))
|
|
|
- {
|
|
|
- return array('offset'=>0, 'data'=>$log_file . 'not exists or not readable');
|
|
|
- }
|
|
|
- // 读文件
|
|
|
- $h = fopen($log_file, 'r');
|
|
|
-
|
|
|
- // 如果有时间，则进行二分查找，加速查询
|
|
|
- if($start_time && $offset === '' && ($file_size = filesize($log_file) > 50000))
|
|
|
- {
|
|
|
- $offset = $this->binarySearch(0, $file_size, $start_time-1, $h);
|
|
|
- $offset = $offset < 1000 ? 0 : $offset - 1000;
|
|
|
- }
|
|
|
-
|
|
|
- // 正则表达式
|
|
|
- $pattern = "/^([\d: \-]+)\t";
|
|
|
-
|
|
|
- if($module && $module != 'WorkerMan')
|
|
|
- {
|
|
|
- $pattern .= $module."::";
|
|
|
- }
|
|
|
- else
|
|
|
- {
|
|
|
- $pattern .= ".*::";
|
|
|
- }
|
|
|
-
|
|
|
- if($interface && $module != 'WorkerMan')
|
|
|
- {
|
|
|
- $pattern .= $interface."\t";
|
|
|
- }
|
|
|
- else
|
|
|
- {
|
|
|
- $pattern .= ".*\t";
|
|
|
- }
|
|
|
-
|
|
|
- if($code !== '')
|
|
|
- {
|
|
|
- $pattern .= "code:$code\t";
|
|
|
- }
|
|
|
- else
|
|
|
- {
|
|
|
- $pattern .= "code:\d+\t";
|
|
|
- }
|
|
|
-
|
|
|
- if($msg)
|
|
|
- {
|
|
|
- $pattern .= "msg:$msg";
|
|
|
- }
|
|
|
-
|
|
|
- $pattern .= '/';
|
|
|
-
|
|
|
- // 指定偏移位置
|
|
|
- if($offset >= 0)
|
|
|
- {
|
|
|
- fseek($h, (int)$offset);
|
|
|
- }
|
|
|
-
|
|
|
- // 查找符合条件的数据
|
|
|
- $now_count = 0;
|
|
|
- $log_buffer = '';
|
|
|
-
|
|
|
- while(1)
|
|
|
- {
|
|
|
- if(feof($h))
|
|
|
- {
|
|
|
- break;
|
|
|
- }
|
|
|
- // 读1行
|
|
|
- $line = fgets($h);
|
|
|
- if(preg_match($pattern, $line, $match))
|
|
|
- {
|
|
|
- // 判断时间是否符合要求
|
|
|
- $time = strtotime($match[1]);
|
|
|
- if($start_time)
|
|
|
- {
|
|
|
- if($time<$start_time)
|
|
|
- {
|
|
|
- continue;
|
|
|
- }
|
|
|
- }
|
|
|
- if($end_time)
|
|
|
- {
|
|
|
- if($time>$end_time)
|
|
|
- {
|
|
|
- break;
|
|
|
- }
|
|
|
- }
|
|
|
- // 收集符合条件的log
|
|
|
- $log_buffer .= $line;
|
|
|
- if(++$now_count >= $count)
|
|
|
- {
|
|
|
- break;
|
|
|
- }
|
|
|
- }
|
|
|
- }
|
|
|
- // 记录偏移位置
|
|
|
- $offset = ftell($h);
|
|
|
- return array('offset'=>$offset, 'data'=>$log_buffer);
|
|
|
- }
|
|
|
- /**
|
|
|
- * 日志二分查找法
|
|
|
- * @param int $start_point
|
|
|
- * @param int $end_point
|
|
|
- * @param int $time
|
|
|
- * @param fd $fd
|
|
|
- * @return int
|
|
|
- */
|
|
|
- protected function binarySearch($start_point, $end_point, $time, $fd)
|
|
|
- {
|
|
|
- // 计算中点
|
|
|
- $mid_point = (int)(($end_point+$start_point)/2);
|
|
|
-
|
|
|
- // 定位文件指针在中点
|
|
|
- fseek($fd, $mid_point);
|
|
|
-
|
|
|
- // 读第一行
|
|
|
- $line = fgets($fd);
|
|
|
- if(feof($fd) || false === $line)
|
|
|
- {
|
|
|
- return ftell($fd);
|
|
|
- }
|
|
|
-
|
|
|
- // 第一行可能数据不全，再读一行
|
|
|
- $line = fgets($fd);
|
|
|
- if(feof($fd) || false === $line || trim($line) == '')
|
|
|
- {
|
|
|
- return ftell($fd);
|
|
|
- }
|
|
|
-
|
|
|
- // 判断是否越界
|
|
|
- $current_point = ftell($fd);
|
|
|
- if($current_point>=$end_point)
|
|
|
- {
|
|
|
- return $end_point;
|
|
|
- }
|
|
|
-
|
|
|
- // 获得时间
|
|
|
- $tmp = explode("\t", $line);
|
|
|
- $tmp_time = strtotime($tmp[0]);
|
|
|
-
|
|
|
- // 判断时间，返回指针位置
|
|
|
- if($tmp_time > $time)
|
|
|
- {
|
|
|
- return $this->binarySearch($start_point, $current_point, $time, $fd);
|
|
|
- }
|
|
|
- elseif($tmp_time < $time)
|
|
|
- {
|
|
|
- return $this->binarySearch($current_point, $end_point, $time, $fd);
|
|
|
- }
|
|
|
- else
|
|
|
- {
|
|
|
- return $current_point;
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- public function tcpSendToClient($str_to_send)
|
|
|
- {
|
|
|
- $send_len = @stream_socket_sendto($this->connections[$this->currentDealFd], $str_to_send);
|
|
|
- if($send_len === strlen($str_to_send))
|
|
|
- {
|
|
|
- return true;
|
|
|
- }
|
|
|
- if($send_len > 0)
|
|
|
- {
|
|
|
- $this->sendBuffers[$this->currentDealFd] = substr($str_to_send, $send_len);
|
|
|
- }
|
|
|
- else
|
|
|
- {
|
|
|
- $this->sendBuffers[$this->currentDealFd] = $str_to_send;
|
|
|
- }
|
|
|
- $this->event->add($this->connections[$this->currentDealFd], \Man\Core\Events\BaseEvent::EV_WRITE, array($this, 'tcpWriteToClient'), array($this->currentDealFd));
|
|
|
- }
|
|
|
}
|
|
|
|
|
|
/**
|
|
|
@@ -716,4 +366,4 @@ class StatisticProtocol
|
|
|
'msg' => $msg,
|
|
|
);
|
|
|
}
|
|
|
-}
|
|
|
+}
|
walkor