stats.go

package service

import (
	"fmt"
	"strings"

	"github.com/imdevlab/tracing/collector/misc"
	"github.com/imdevlab/tracing/pkg/constant"
	"github.com/imdevlab/tracing/pkg/stats"

	"github.com/imdevlab/tracing/pkg/pinpoint/thrift/trace"
)

// Stats 计算结果
type Stats struct {
	httpCodes map[int32]struct{} // http标准code
	api       *stats.ApiStore    // api统计
	sql       *stats.SQLS        // sql语句计算统计
	method    *stats.Methods     // 接口计算统计
	exception *stats.Exceptions  // 异常计算统计
	apiMap    *stats.ApiMap      // 接口拓扑图
	srvMap    *stats.SrvMap      // 应用拓扑图
}

// NewStats ....
func NewStats(httpCodes, customCodes map[int32]struct{}) *Stats {
	stats := &Stats{
		httpCodes: make(map[int32]struct{}),
		api:       stats.NewApiStore(),
		sql:       stats.NewSQLS(),
		method:    stats.NewMethods(),
		exception: stats.NewExceptions(),
		apiMap:    stats.NewApiMap(),
		srvMap:    stats.NewSrvMap(),
	}

	// http codes默认给200
	stats.httpCodes[200] = struct{}{}

	// 添加策略
	for code := range httpCodes {
		stats.httpCodes[code] = struct{}{}
	}
	return stats
}

// 如果该应用为父节点，那么要统计自己的api，目标为自己，请求者为unknow
func (s *Stats) selfApiCounter(span *trace.TSpan) {
	// 查找应用
	app, ok := s.api.Apps[span.GetApplicationName()]
	if !ok {
		app = stats.NewApp()
		s.api.Apps[span.GetApplicationName()] = app
	}

	url, ok := app.Urls[span.GetRPC()]
	if !ok {
		url = stats.NewUrl()
		app.Urls[span.GetRPC()] = url
	}

	url.Duration += span.GetElapsed()
	url.AccessCount++
	if span.GetElapsed() > url.MaxDuration {
		url.MaxDuration = span.GetElapsed()
	}

	if url.MinDuration == 0 || url.MinDuration > span.GetElapsed() {
		url.MinDuration = span.GetElapsed()
	}

	cacheCode := false
	var code int32

	// Annotations
	for _, annotation := range span.GetAnnotations() {
		if annotation.GetKey() == constant.HTTP_STATUS_CODE {
			cacheCode = true
			code = annotation.GetValue().GetIntValue()
			break
		}
	}

	// 检查code是否正确, 如果在策略里面没找到发现code，那么说明这次http请求失败
	if cacheCode {
		if _, ok := s.httpCodes[code]; !ok {
			url.AccessCodeErrCount++
		}
	}

	// 耗时小于满意时间满意次数加1
	if span.GetElapsed() < misc.Conf.Stats.SatisfactionTime {
		url.SatisfactionCount++
		// 耗时小于可容忍时间，可容忍次数加一， 其他都为沮丧次数
	} else if span.GetElapsed() < misc.Conf.Stats.TolerateTime {
		url.TolerateCount++
	}

}

// SpanCounter 计算
func (s *Stats) SpanCounter(span *trace.TSpan) error {
	// 计算API信息
	for _, event := range span.GetSpanEventList() {
		s.eventCounter(span.GetApplicationName(), span.GetRPC(), event)
	}

	// 如果该应用为父节点，那么要统计自己的api，目标为自己，请求者为unknow
	if span.GetParentSpanId() == -1 {
		s.selfApiCounter(span)
	}

	// 计算API被哪些服务调用
	{
		s.apiMapCounter(span)
	}

	// 计算服务拓扑图
	{
		s.parentMapCounter(span)
	}

	return nil
}

// SpanChunkCounter counter 计算
func (s *Stats) SpanChunkCounter(spanChunk *trace.TSpanChunk) error {
	for _, event := range spanChunk.GetSpanEventList() {
		s.eventCounter(spanChunk.GetApplicationName(), "", event)
	}
	return nil
}

// sqlCount 计算sql
func (s *Stats) sqlCount(event *trace.TSpanEvent) {
	var sqlID int32
	cacheSQLID := false
	for _, annotation := range event.GetAnnotations() {
		if annotation.GetKey() == constant.SQL_ID {
			sqlID = annotation.Value.GetIntStringStringValue().GetIntValue()
			cacheSQLID = true
			break
		}
	}
	// 没有sqlID 直接返回
	if !cacheSQLID {
		logger.Warn("unfind sql id")
		return
	}

	sql, ok := s.sql.Get(sqlID)
	if !ok {
		sql = stats.NewSQL()
		s.sql.Store(sqlID, sql)
	}

	sql.Duration += event.GetEndElapsed()
	sql.Count++

	if event.GetEndElapsed() > sql.MaxDuration {
		sql.MaxDuration = event.GetEndElapsed()
	}

	if sql.MinDuration == 0 || sql.MinDuration > event.GetEndElapsed() {
		sql.MinDuration = event.GetEndElapsed()
	}

	// 是否有异常抛出
	if event.GetExceptionInfo() != nil {
		sql.ErrCount++
	}
}

// apiMapCounter 接口被哪些服务调用计算
func (s *Stats) apiMapCounter(span *trace.TSpan) {
	apiStr := span.GetRPC()
	if len(apiStr) <= 0 {
		return
	}

	api, ok := s.apiMap.Apis[apiStr] //.APIS[]
	if !ok {
		api = stats.NewApi()
		s.apiMap.Apis[apiStr] = api
	}

	var parentName string
	var parentType int16
	// spanID 为-1的情况该服务就是父节点，查不到被谁调用，这里可以考虑能不能抓到请求者到IP
	if span.ParentSpanId == -1 {
		parentName = "UNKNOWN"
		parentType = constant.SERVERTYPE_UNKNOWN
	} else {
		parentName = span.GetParentApplicationName()
		parentType = span.GetParentApplicationType()
	}

	parents, ok := api.Parents[parentName]
	if !ok {
		parents = stats.NewParent()
		parents.Type = parentType
		api.Parents[parentName] = parents
	}

	parents.AccessCount++
	parents.AccessDuration += span.Elapsed
	if span.GetErr() != 0 {
		parents.ExceptionCount++
	}
}

// parentMapCounter 计算服务拓扑图
func (s *Stats) parentMapCounter(span *trace.TSpan) {
	// spanID 为-1的情况该服务就是父节点，请求者应该是没接入监控
	if span.ParentSpanId == -1 {
		s.srvMap.UnknowParent.AccessCount++
		s.srvMap.UnknowParent.AccessDuration += span.GetElapsed()
		return
	}

	s.srvMap.AppType = span.GetServiceType()
	parent, ok := s.srvMap.Parents[span.GetParentApplicationName()]
	if !ok {
		parent = stats.NewSrvParent()
		parent.Type = span.GetParentApplicationType()
		s.srvMap.Parents[span.GetParentApplicationName()] = parent
	}

	parent.TargetCount++
	if span.GetErr() != 0 {
		parent.TargetErrCount++
	}
}

// isDubbo 检查event类型是否为rpc接口
func isDubbo(eventType int16) bool {
	if eventType == constant.DUBBO_PROVIDER ||
		eventType == constant.DUBBO_CONSUMER ||
		eventType == constant.DUBBO {
		return true
	}
	return false
}

// isHttp 检查event类型是否为http接口
func isHttp(eventType int16) bool {
	if eventType == constant.HTTP_CLIENT_3 ||
		eventType == constant.HTTP_CLIENT_3_INTERNAL ||
		eventType == constant.HTTP_CLIENT_4 ||
		eventType == constant.HTTP_CLIENT_4_INTERNAL {
		return true
	}
	return false
}

// isDB 检查event类型是否为数据库操作
func isDB(eventType int16) bool {
	if eventType == constant.MYSQL_EXECUTE_QUERY ||
		eventType == constant.ORACLE_EXECUTE_QUERY ||
		eventType == constant.MARIADB_EXECUTE_QUERY {
		return true
	}
	return false
}

func getip(destinationID string) (string, error) {
	strs := strings.Split(destinationID, ":")
	if len(strs) != 2 {
		return "", fmt.Errorf("unknow addr")
	}
	if len(strs[0]) == 0 {
		return "", fmt.Errorf("error ip")
	}
	return strs[0], nil
}

func gethost(destinationID string) (string, error) {
	strs := strings.Split(destinationID, ":")
	if len(strs) != 2 {
		return "", fmt.Errorf("unknow addr")
	}
	if len(strs[0]) == 0 {
		return "", fmt.Errorf("error ip")
	}
	return strs[0], nil
}

func (s *Stats) urlCounter(event *trace.TSpanEvent) {
	// 是否已经缓存api
	cacheApi := false
	cacheCode := false
	var code int32
	var urlStr string

	// Annotations
	for _, annotation := range event.GetAnnotations() {
		if annotation.GetKey() == constant.HTTP_URL {
			cacheApi = true
			urlStr = annotation.GetValue().GetStringValue()
			continue
		}
		if annotation.GetKey() == constant.HTTP_STATUS_CODE {
			cacheCode = true
			code = annotation.GetValue().GetIntValue()
		}
		if cacheApi && cacheCode {
			break
		}
	}

	if cacheApi {
		// 获取app信息
		destinationID := event.GetDestinationId()
		if len(destinationID) <= 0 {
			return
		}

		// 通过IP获取app name
		ip, err := getip(destinationID)
		if err == nil {
			appName, ok := gCollector.apps.getNameByIP(ip)
			if ok {
				destinationID = appName
			}
		} else {
			// 如果不是IP尝试切一下-，如果还是找不到那么就使用destinationID
			destinationID = cutVip(destinationID)
		}

		// 查找应用
		app, ok := s.api.Apps[destinationID]
		if !ok {
			app = stats.NewApp()
			s.api.Apps[destinationID] = app
		}

		url, ok := app.Urls[urlStr]
		if !ok {
			url = stats.NewUrl()
			app.Urls[urlStr] = url
		}

		url.Duration += event.GetEndElapsed()
		url.AccessCount++
		if event.GetEndElapsed() > url.MaxDuration {
			url.MaxDuration = event.GetEndElapsed()
		}

		if url.MinDuration == 0 || url.MinDuration > event.GetEndElapsed() {
			url.MinDuration = event.GetEndElapsed()
		}
		// 检查code是否正确, 如果在策略里面没找到发现code，那么说明这次http请求失败
		if cacheCode {
			if _, ok := s.httpCodes[code]; !ok {
				url.AccessCodeErrCount++
			}
		}
		// 检查网络错误
		// AccessNetErrCount  int   // 访问网络错误数

		// 耗时小于满意时间满意次数加1
		if event.GetEndElapsed() < misc.Conf.Stats.SatisfactionTime {
			url.SatisfactionCount++
			// 耗时小于可容忍时间，可容忍次数加一， 其他都为沮丧次数
		} else if event.GetEndElapsed() < misc.Conf.Stats.TolerateTime {
			url.TolerateCount++
		}
	}
}

func (s *Stats) dubboCounter(appName string, event *trace.TSpanEvent) {
	for _, annotation := range event.GetAnnotations() {
		if annotation.GetKey() == constant.DUBBO_RPC {
			app, ok := s.api.Apps[appName]
			if !ok {
				app = stats.NewApp()
				s.api.Apps[appName] = app
			}
			dubbo, ok := app.Dubbos[annotation.GetValue().GetStringValue()]
			if !ok {
				dubbo = stats.NewDubbo()
				app.Dubbos[annotation.GetValue().GetStringValue()] = dubbo
			}
			dubbo.AccessCount++
			dubbo.Duration += event.GetEndElapsed()

			if dubbo.MinDuration == 0 || dubbo.MinDuration > event.GetEndElapsed() {
				dubbo.MinDuration = event.GetEndElapsed()
			}

			if event.GetEndElapsed() > dubbo.MaxDuration {
				dubbo.MaxDuration = event.GetEndElapsed()
			}

			if event.GetExceptionInfo() == nil {
				dubbo.AccessExErrCount++
			}

			break
		}
	}
}

// eventCounter event数据统计 计算api信息， sql、method、异常等
func (s *Stats) eventCounter(appName, apiStr string, event *trace.TSpanEvent) {
	if isDubbo(event.GetServiceType()) {
		// dubbo
		s.dubboCounter(appName, event)
	} else if isHttp(event.GetServiceType()) {
		// http 统计
		s.urlCounter(event)
	} else if isDB(event.GetServiceType()) {
		// 数据库统计
		s.sqlCount(event)
	}

	// method统计
	s.methodCount(apiStr, event)

	// exception
	s.exceptionCount(event)

	// app后续服务拓扑图计算
	s.targetMapCounter(event)
}

// exceptionCount 异常统计
func (s *Stats) exceptionCount(event *trace.TSpanEvent) {
	// 参看是否存在异常，不存在直接返回
	exInfo := event.GetExceptionInfo()
	if exInfo == nil {
		return
	}

	method, ok := s.exception.Get(event.GetApiId())
	if !ok {
		method = stats.NewExMethod()
		s.exception.Store(event.GetApiId(), method)
	}

	ex, ok := method.Get(exInfo.GetIntValue())
	if !ok {
		ex = stats.NewException()
		method.Store(exInfo.GetIntValue(), ex)
	}

	ex.Duration += event.GetEndElapsed()
	ex.Type = int(event.GetServiceType())
	ex.Count++

	if event.GetEndElapsed() > ex.MaxDuration {
		ex.MaxDuration = event.GetEndElapsed()
	}

	if ex.MinDuration == 0 || ex.MinDuration > event.GetEndElapsed() {
		ex.MinDuration = event.GetEndElapsed()
	}
}

func (s *Stats) methodCount(apiStr string, event *trace.TSpanEvent) {
	// 保存api到methods
	if len(apiStr) > 0 {
		if len(s.method.ApiStr) == 0 {
			s.method.ApiStr = apiStr
		}
	}
	method, ok := s.method.Get(event.GetApiId())
	if !ok {
		method = stats.NewMethod(event.GetServiceType())
		s.method.Store(event.GetApiId(), method)
	}

	method.Duration += event.GetEndElapsed()
	method.Count++

	if event.GetEndElapsed() > method.MaxDuration {
		method.MaxDuration = event.GetEndElapsed()
	}

	if method.MinDuration == 0 || method.MinDuration > event.GetEndElapsed() {
		method.MinDuration = event.GetEndElapsed()
	}

	// 是否有异常抛出
	if event.GetExceptionInfo() != nil {
		method.ErrCount++
	}
}

func cutVip(destination string) string {
	var name string
	names := strings.Split(destination, "-")
	if len(names) == 1 {
		name = destination
	} else if len(names) == 3 {
		name = names[1]
	} else if len(names) == 4 {
		name = names[1] + names[2]
	}
	return name
}

// 计算child拓扑图
func (s *Stats) targetMapCounter(event *trace.TSpanEvent) {

	if event.ServiceType == constant.DUBBO_CONSUMER ||
		event.ServiceType == constant.HTTP_CLIENT_3 ||
		event.ServiceType == constant.HTTP_CLIENT_3_INTERNAL ||
		event.ServiceType == constant.HTTP_CLIENT_4 ||
		event.ServiceType == constant.HTTP_CLIENT_4_INTERNAL ||
		event.ServiceType == constant.MYSQL_EXECUTE_QUERY ||
		event.ServiceType == constant.REDIS ||
		event.ServiceType == constant.ORACLE_EXECUTE_QUERY ||
		event.ServiceType == constant.MARIADB_EXECUTE_QUERY {
		destinationID := event.GetDestinationId()
		if len(destinationID) <= 0 {
			return
		}

		targets, ok := s.srvMap.Targets[event.ServiceType]
		if !ok {
			targets = make(map[string]*stats.Target)
			s.srvMap.Targets[event.ServiceType] = targets
		}

		// http&&dubbo做特殊处理
		if event.ServiceType == constant.HTTP_CLIENT_3 ||
			event.ServiceType == constant.HTTP_CLIENT_3_INTERNAL ||
			event.ServiceType == constant.HTTP_CLIENT_4 ||
			event.ServiceType == constant.HTTP_CLIENT_4_INTERNAL ||
			event.ServiceType == constant.DUBBO_CONSUMER {
			ip, err := getip(destinationID)
			if err == nil {
				appName, ok := gCollector.apps.getNameByIP(ip)
				if ok {
					destinationID = appName
				}
			} else {
				// 如果不是IP尝试切一下-，如果还是找不到那么就使用destinationID
				destinationID = cutVip(destinationID)
			}
		}

		target, ok := targets[destinationID]
		if !ok {
			target = stats.NewTarget()
			targets[destinationID] = target
		}

		// http code统计
		if event.ServiceType == constant.HTTP_CLIENT_3 ||
			event.ServiceType == constant.HTTP_CLIENT_3_INTERNAL ||
			event.ServiceType == constant.HTTP_CLIENT_4 ||
			event.ServiceType == constant.HTTP_CLIENT_4_INTERNAL {
			for _, annotation := range event.GetAnnotations() {
				if annotation.GetKey() == constant.HTTP_STATUS_CODE {
					if _, ok := s.httpCodes[annotation.Value.GetIntValue()]; !ok {
						target.AccessErrCount++
					}
				}
			}
		}
		target.AccessCount++
		target.AccessDuration += event.EndElapsed
	}
}