light-apiengine-feature/middleware/logger/cassandra.go

package logger

import (
	"encoding/json"
	"strconv"
	"strings"
	"time"

	"git.qianqiusoft.com/public/glog"
	"git.qianqiusoft.com/qianqiusoft/light-apiengine/config"
	"github.com/gocql/gocql"
)

// cassandra 配置
var _clusterCfg *gocql.ClusterConfig = nil

func init() {
	// 设置cassandar 配置
	cassandra := config.AppConfig.GetKey("cassandra")
	keyspace := config.AppConfig.GetKey("keyspace")
	_clusterCfg = gocql.NewCluster(strings.Split(cassandra, ",")...)
	_clusterCfg.Keyspace = keyspace
	_clusterCfg.Consistency = gocql.Quorum
	//设置连接池的数量,默认是2个（针对每一个host,都建立起NumConns个连接）
	_clusterCfg.NumConns = 3

	session, err := _clusterCfg.CreateSession()
	if err != nil {
		glog.Errorln("create session err", err.Error())
		return
	}
	time.Sleep(1 * time.Second) //Sleep so the fillPool can complete.
	defer session.Close()
}

// cassandra 日志 处理接口
func cassandraLogHandler(logInfo *LogInfo) {
	session, err := _clusterCfg.CreateSession()
	if err != nil {
		glog.Errorln("创建cassandra session错误", err.Error())
		return
	}
	defer session.Close()

	var respBodyObj struct {
		Code int
		Msg  string
		Data interface{}
	}
	if logInfo.RespBody != "" {
		err = json.Unmarshal([]byte(logInfo.RespBody), &respBodyObj)
		if err != nil {
			glog.Errorln("json.Unmarshal 错误", err.Error())
		} else {
			logInfo.RespCode = strconv.Itoa(respBodyObj.Code)
			logInfo.RespMsg = respBodyObj.Msg
			if respBodyObj.Data != nil {
				str, err := json.Marshal(respBodyObj.Data)
				if err == nil {
					logInfo.RespData = string(str)
				}
			}
		}
	}
	logInfo.CostTime = float64(logInfo.RespTime.Sub(logInfo.ReqTime).Seconds())

	cqlformat := `insert into 
		log_info(app_name, req_date, req_time, login_id, req_method, req_url, req_proto, req_ua, req_referer, req_post_data, resp_time, resp_body, resp_code, resp_msg, resp_data, cost_time) 
		values(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?);`
	reqDate := logInfo.ReqTime.Format("20060102")
	//q := session.Query(cqlformat, __appName, reqDate, logInfo.ReqTime.Format("2006-01-02 15:04:05"), logInfo.ReqMethod, logInfo.ReqUrl, logInfo.ReqProto, logInfo.ReqUa, logInfo.ReqReferer, logInfo.ReqPostData, logInfo.RespTime.Format("2006-01-02 15:04:05"), logInfo.RespBody, logInfo.RespCode, logInfo.RespMsg, logInfo.RespData, logInfo.CostTime)
	batch := session.NewBatch(gocql.UnloggedBatch) // 应该批量插入多条，例如20条
	batch.Query(cqlformat, __appName, reqDate, logInfo.ReqTime.Format("2006-01-02 15:04:05"), logInfo.LoginId, logInfo.ReqMethod, logInfo.ReqUrl, logInfo.ReqProto, logInfo.ReqUa, logInfo.ReqReferer, strings.TrimSpace(logInfo.ReqPostData), logInfo.RespTime.Format("2006-01-02 15:04:05"), strings.TrimSpace(logInfo.RespBody), logInfo.RespCode, strings.TrimSpace(logInfo.RespMsg), strings.TrimSpace(logInfo.RespData), logInfo.CostTime)
	if err := session.ExecuteBatch(batch); err != nil {
		glog.Errorln("批量插入日志错误:", err)
		return
	}
}

func GetLogByCassandra(appName, date, beginTime, endTime, loginId string) ([]map[string]interface{}, error) {
	session, err := _clusterCfg.CreateSession()
	if err != nil {
		glog.Errorln("创建cassandra session错误", err.Error())
		return nil, err
	}
	defer session.Close()

	params := []interface{}{appName, date, beginTime, endTime}

	cqlformat := `select app_name, req_date, req_time, login_id, req_method, req_url, req_proto, req_ua, req_referer, req_post_data, resp_time, resp_body, resp_code, resp_msg, resp_data, cost_time
			from log_info where app_name = ? and req_date = ? and req_time >= ? and req_time <= ?
			`
	if loginId != "" {
		cqlformat += " and login_id = ?"
		params = append(params, loginId)
	}
	var appName1 string
	var reqDate string
	var reqTime, loginId1, reqMethod, reqUrl, reqProto, reqUa, reqReferer, reqPostData, respTime, respBody, respCode, respMsg, respData string
	var costTime float64

	results := make([]map[string]interface{}, 0)
	iter := session.Query(cqlformat, params...).Iter()
	for iter.Scan(&appName1, &reqDate, &reqTime, &loginId1, &reqMethod, &reqUrl, &reqProto, &reqUa, &reqReferer, &reqPostData, &respTime, &respBody, &respCode, &respMsg, &respData, &costTime) {
		result := make(map[string]interface{})
		result["app_name"] = appName1
		result["req_date"] = reqDate
		result["req_time"] = reqTime
		result["login_id"] = loginId1
		result["req_method"] = reqMethod
		result["req_url"] = reqUrl
		result["req_proto"] = reqProto
		result["req_ua"] = reqUa
		result["req_referer"] = reqReferer
		result["req_post_data"] = reqPostData
		result["resp_time"] = respTime
		result["resp_body"] = respBody
		result["resp_code"] = respCode
		result["resp_msg"] = respMsg
		result["resp_data"] = respData
		result["cost_time"] = costTime
		results = append(results, result)
	}
	if err := iter.Close(); err != nil {
		glog.Errorln("iter close err", err.Error())
	}

	return results, nil
}