observe-server/common/opentelemetry/tools.go

package opentelemetry

import (
	"fmt"
	"net/url"
	"strconv"
	"strings"

	"github.com/pkg/errors"
	"go.opentelemetry.io/collector/pdata/pcommon"
	conventions "go.opentelemetry.io/collector/semconv/v1.18.0"
)

type SpanAttributes struct {
	mp pcommon.Map
}

func NewSpanAttributes(attrs map[string]string) SpanAttributes {
	mp := pcommon.NewMap()
	for key, val := range attrs {
		mp.PutStr(key, val)
	}
	return SpanAttributes{mp}
}

func NewSpanAttributesWithRaw(attrs pcommon.Map) SpanAttributes {
	return SpanAttributes{mp: attrs}
}

func (sa SpanAttributes) getAttribute(keys ...string) (val pcommon.Value) {
	for _, key := range keys {
		if v, ok := sa.mp.Get(key); ok {
			return v
		}
		// log.Debugf("获取span属性%s失败", key)
	}
	return pcommon.NewValueBytes()
}

func (sa SpanAttributes) HttpRoute() (route string) {
	route = sa.getAttribute("http.route", "http.target").AsString()
	if route == "" {
		rawUrl := sa.HttpUrl()
		if urlInfo, err := url.Parse(rawUrl); err == nil {
			route = urlInfo.Path
		}
	}
	routes := strings.Split(route, "?")
	if len(routes) > 1 {
		route = routes[0]
	}
	return
}

func (sa SpanAttributes) HttpTarget() (target string) {
	target = sa.getAttribute("http.target", "http.route").AsString()
	if target == "" {
		rawUrl := sa.HttpUrl()
		if urlInfo, err := url.Parse(rawUrl); err == nil {
			if urlInfo.Path != "" && urlInfo.RawQuery != "" {
				target = fmt.Sprintf("%s?%s", urlInfo.Path, urlInfo.RawQuery)
			} else if urlInfo.Path != "" {
				target = urlInfo.Path
			}
		}
	}
	return
}

func (sa SpanAttributes) HttpHost() (host string) {
	host = sa.getAttribute("http.host").AsString()
	if host == "" {
		rawUrl := sa.HttpUrl()
		if urlInfo, err := url.Parse(rawUrl); err == nil {
			host = urlInfo.Host
		}
	}
	return
}

// getHttpUrl 大部分都有http.url，但少部分没有，需要手动拼凑
func (sa SpanAttributes) HttpUrl() (url string) {
	url = sa.getAttribute("http.url", "url.full").AsString()
	schema := sa.getAttribute("http.schema", "url.scheme").AsString()
	if schema == "" {
		schema = "http"
	}
	if url == "" {
		host := sa.getAttribute("http.host").AsString()
		if host == "" { // 没有host，返回空url
			return
		}
		target := sa.getAttribute("http.target").AsString()
		url = fmt.Sprintf("%s://%s%s", schema, host, target)
	} else {
		if strings.HasPrefix(url, "http://") || strings.HasPrefix(url, "https://") {
			return
		}
		if strings.HasPrefix(url, "/") { // 有sdk生成的http.url仅包抱一个类似于/path的结构，对于该种结构，对于这种结构，直接返回
			return
		}
		url = schema + "://" + url // 之所以加一个http://前缀，是因为url.Parse时，如果没有http://前缀时，不能正常解析
	}
	return
}

func (sa SpanAttributes) StatusCode() int64 {
	code := sa.getAttribute("http.status_code", "http.response.status_code").AsString()
	statusCode, err := strconv.ParseInt(code, 10, 64)
	if err != nil {
		return -1
	}
	return statusCode
}

func (sa SpanAttributes) HttpFlavor() string {
	flavor := sa.getAttribute("http.flavor", "http.protocol").AsString()
	if arr := strings.Split(flavor, "/"); len(arr) > 1 {
		return arr[1]
	}
	return flavor
}

func (sa SpanAttributes) HttpMethod() string {
	// http.request.method为新版本定义
	// http.method为旧版本定义
	return sa.getAttribute("http.request.method", "http.method").AsString()
}

func (sa SpanAttributes) UserAgent() string {
	return sa.getAttribute("http.user_agent", "user_agent.original").AsString()
}

func (sa SpanAttributes) DbSystem() string {
	return sa.getAttribute("db.system").AsString()
}

func (sa SpanAttributes) IsDB() bool {
	return sa.getAttribute("db.system", "db.type", "db.operate").AsString() != ""
}

// 获取数据库服务的地址
func (sa SpanAttributes) DbServerAddress() string {
	return sa.getAttribute("server.address", "network.peer.address", "net.peer.name").AsString()
}

// 获取数据库服务的端口
func (sa SpanAttributes) DbServerPort() string {
	return sa.getAttribute("server.port", "network.peer.port", "net.peer.port").AsString()
}

func (sa SpanAttributes) NetworkPeerAddress() string {
	// server.address代表真实的服务的地址，即使中间有代理，这个值也是真实服务的值，大根率没有
	// 后面两个都代表直接通信的地址，可能是代理，第一个为目前官方定义的，第二个是之前版本定义的
	return sa.getAttribute("server.address", "network.peer.address", "net.peer.name").AsString()
}
func (sa SpanAttributes) NetworkPeerPort() string {
	// 参考NetwordPeerAddress说明
	return sa.getAttribute("server.port", "network.peer.port", "net.peer.port").AsString()
}

func (sa SpanAttributes) RpcSystem() string {
	return sa.getAttribute("rpc.system").AsString()
}

func (sa SpanAttributes) WrapSpanName(spanName, spanKind string) string {
	newSpanName := spanName
	if spanKind == "SPAN_KIND_CLIENT" {
		route := sa.HttpRoute()
		if route != "" {
			newSpanName = fmt.Sprintf("%s -> %s", spanName, route)
		}
	}
	return newSpanName
}

// 返回请求方法和spanName
func (sa SpanAttributes) UnWrapSpanName(spanName string) (string, string, error) {
	if strings.Contains(spanName, " -> ") {
		arr := strings.Split(spanName, " -> ")
		method, route := arr[0], arr[1]
		if strings.HasPrefix(method, "HTTP ") {
			httpMethod := strings.Split(method, " ")
			method = httpMethod[1]
		}
		return method, route, nil
	}
	return "", "", fmt.Errorf("span name: %s 未包装过", spanName)
}

// 消息系统 ------------- start ---------------
func (sa SpanAttributes) MessagingSystem() string {
	return sa.getAttribute("messaging.system").AsString()
}

func (sa SpanAttributes) MessagingOperation() string {
	return sa.getAttribute("messaging.operation").AsString()
}

func (sa SpanAttributes) MessagingServerAddress() string {
	return sa.getAttribute("server.address", "network.peer.address", "net.peer.name").AsString()
}

// 消息系统 ------------- end ---------------

type ResourceAttributes struct {
	res *pcommon.Resource
}

func NewResourceAttributes(res *pcommon.Resource) ResourceAttributes {
	return ResourceAttributes{res}
}

// 获取app name，虽然这里叫app name，但其实是app alias
func (ra *ResourceAttributes) AppName() (string, error) {
	res := ra.res
	// 在collector未使用res app.name作为AppAlias时, 暂时注释
	// if appName, ok := res.Attributes().Get("app.name"); ok { // 优先使用 resource attributes中的 app.name参数判断
	// 	return appName.AsString(), nil
	// }
	var err error
	appAlias := ""
	if command, ok := res.Attributes().Get(conventions.AttributeProcessCommandLine); ok {
		cmds := strings.Split(command.AsString(), " ")
		for _, cmd := range cmds {
			if strings.HasPrefix(cmd, "-DAPP_NAME") { // 仅针对java
				arr := strings.Split(cmd, "=")
				if len(arr) == 2 {
					appAlias = arr[1]
				} else {
					err = fmt.Errorf("-DAPP_NAME参数非法:%s", cmd)
				}
				break
			}
		}
	}
	if appAlias == "" {
		if appName, ok := res.Attributes().Get("service.namespace"); ok {
			appAlias = appName.AsString()
		} else if appName, ok := res.Attributes().Get("app.name"); ok {
			appAlias = appName.AsString()
		} else {
			appAlias = "UNSET"
			err = errors.New("未设置service.namespace或app.name")
		}
	}
	return appAlias, err
}

// 获取service name
func (ra *ResourceAttributes) ServiceName() (string, error) {
	var serviceName string
	var err error
	if v, ok := ra.res.Attributes().Get(conventions.AttributeServiceName); ok {
		serviceName = v.Str()
	} else {
		err = errors.New("未获取到service name")
	}
	return serviceName, err
}