ntonpower/app/library/LogToKafka.php

<?php

namespace app\library;

use think\facade\Cache;
use think\facade\Env;

/**
 * 日志消息队列
 */
trait LogToKafka
{
    /**
     * 日志放Kafka接口配置
     * @var array
     */
    public static $sendLogToKafkaConfig = [];

    /**
     * 日志类型
     * action_logs  行为日志内容（值为二维数组序列化）（不是本日志传空值）
     * api_logs     接口日志内容（值为二维数组序列化）（不是本日志传空值）
     * lg_logs_view 访客日志内容（值为二维数组序列化）（不是本日志传空值）
     * lg_task_queue队列日志内容（值为二维数组序列化）（不是本日志传空值）
     *
     * @var string[]
     */
    protected static $logType = ['action_logs','api_logs','lg_logs_view','lg_task_queue'];

    /**
     * 日志写入类型，add添加日志，api调用接口，queue队列
     * @var string[]
     */
    protected static $methodType = ['add','api','queue'];

    /**
     * 初始化配置
     * @return array
     */
    private static function initLogConfig():array
    {
        return self::$sendLogToKafkaConfig = [
            'LogsSSOUrl' => Env::get('SSO.UC_URL'),
            'apiRoute'  =>  '/journal/v1/kafkaproducer/send',
            'signStr'   =>  'logsmonitor',
            'secretKey' =>  'logsmonitorscretlist',
            'dbName'    =>  'users'
        ];
    }

    /**
     * 发送请求，生成到api日志kafka
     * @param array $data 请求数据
     * @param string $logType 日志类型
     * @param string $database 操作数据库
     * @param string $dotype 调用接口方式
     * @return bool
     */
    public static function sendLogsToKafkaProduce(array $data, string $logType, string $database, string $dotype):bool
    {
        if (empty(self::$sendLogToKafkaConfig)) {
            self::initLogConfig();
        }
        $newData = [
            'action_logs'   => "",
            'api_logs'      => "",
            'lg_logs_view'  => "",
            'lg_task_queue' => ""
        ];
        //先构建好数据
        if ($logType != '' && in_array($logType,self::$logType)) {
            $newData[$logType] = json_encode($data);
        }
        //构建数据库
        if ($database) {
            $newData['database'] = $database;
        }
        //构建写入kafka的类型，
        if ($dotype != '' && in_array($dotype,self::$methodType)) {
            $newData['dotype'] = $dotype;
        }
        //拿到数据，构建签名
        $newData['sign'] = self::getSign($newData);
        //构建请求接口
        $url = self::$sendLogToKafkaConfig['LogsSSOUrl'].self::$sendLogToKafkaConfig['apiRoute'];
        $result = self::http($url,$newData,true);
        Cache::set('$url',$url,300);
        Cache::set('result',$result,300);
        if(isset($result['errno']) && $result['errno'] == 0) {
            return true;
        }
        return false;
    }

    /**
     * 签名
     * @param array $param 用于验证数据完整性
     * @return string
     */
    public static function getSign(array $param):string
    {
        ksort($param);
        $signStr = self::$sendLogToKafkaConfig['signStr'];
        $secretKey = self::$sendLogToKafkaConfig['secretKey'];
        foreach ( $param as $key => $value ) {
            $signStr = $signStr . $key . "=" . $value . "&";
        }
        $signStr = substr($signStr, 0, -1);
        return base64_encode(hash_hmac("sha1", $signStr, $secretKey, true));
    }


    /**
     * curl请求
     * @param $url  string 请求的url链接
     * @param $data string|array|mixed 请求的数据
     * @param bool $is_post 是否是post请求，默认false
     * @param array $options 是否附带请求头
     * @return array|mixed
     */
    private static function http(string $url, array $data, bool $is_post=false, array $options=[]):array
    {
        $data  = json_encode($data);
        $headerArray = [
            'Content-type: application/json;charset=utf-8',
            'Accept: application/json'
        ];
        $curl = curl_init();
        $arr = [];
        array_push($arr,$url);
        curl_setopt($curl, CURLOPT_URL, $url);
        curl_setopt($curl, CURLOPT_SSL_VERIFYPEER, FALSE);
        curl_setopt($curl, CURLOPT_SSL_VERIFYHOST,FALSE);
        if ($is_post) {
            curl_setopt($curl, CURLOPT_POST, 1);
            curl_setopt($curl, CURLOPT_POSTFIELDS, $data);
        }
        curl_setopt($curl,CURLOPT_HTTPHEADER,$headerArray);
        curl_setopt($curl, CURLOPT_RETURNTRANSFER, 1);
        if (!empty($options['cookie'])) {
            curl_setopt($curl, CURLOPT_COOKIE, $options['cookie']);
        }
        $output = curl_exec($curl);
        $http_status = curl_errno($curl);
        $http_msg = curl_error($curl);
        curl_close($curl);
        if ($http_status == 0) {
            return json_decode($output, true);
        } else {
            return ['status' => $http_status, 'message' => $http_msg, 'data' => []];
        }
    }

}