observe-server/cmd/consumer/metrics/observe_server_duration.go

package metrics

import (
	"context"
	"go-admin/app/observe/models/query"
	"go-admin/common/opentelemetry"
	"strings"
	"sync"

	log "github.com/go-admin-team/go-admin-core/logger"
	"github.com/go-admin-team/go-admin-core/sdk/config"
	"go.opentelemetry.io/collector/pdata/pcommon"
	"go.opentelemetry.io/collector/pdata/ptrace"
	conventions "go.opentelemetry.io/collector/semconv/v1.18.0"
	"go.opentelemetry.io/otel"
	"go.opentelemetry.io/otel/attribute"
	"go.opentelemetry.io/otel/metric"
)

type ObserveServerDuration struct{ metric.Float64Histogram }

func NewObserveServerDuration(name string) ObserveServerDuration {

	provider := otel.GetMeterProvider()
	meter := provider.Meter("observe.consumer.metrics")
	if h, err := meter.Float64Histogram(
		"observe.server.duration",
		metric.WithDescription("observe服务端处理时延统计"),
		metric.WithUnit("ms"),
	); err != nil {
		panic(err)
	} else {
		return ObserveServerDuration{h}
	}
}

func (h ObserveServerDuration) Record(wg *sync.WaitGroup, ch <-chan struct{}, span ptrace.Span, resource pcommon.Resource) {
	defer wg.Done()
	defer func() {
		<-ch
	}()
	if span.Kind() != ptrace.SpanKindServer {
		return
	}
	resAttrs := opentelemetry.NewResourceAttributes(&resource)
	spanAttrs := opentelemetry.NewSpanAttributesWithRaw(span.Attributes())
	appAlias, _ := resAttrs.AppName()
	if appAlias == "UNSET" && config.ApplicationConfig.Mode == "prod" { // 为了降低数据量， 客户生产暂时不再捕获未设置AppAlias的数据
		return
	}
	duration := span.EndTimestamp().AsTime().Sub(span.StartTimestamp().AsTime()).Milliseconds()
	appId, err := query.NewApp().Alias2ID(appAlias)
	if err != nil || appId == 0 {
		log.Errorf("获取appId失败: %s %s", appAlias, err)
		return
	}
	serviceName, ok := resource.Attributes().Get(conventions.AttributeServiceName)
	if !ok {
		log.Errorf("获取service name失败: %s %s", appAlias, span.SpanID().String())
		return
	}
	urlMappingQuery := query.NewUrlMapping()
	// kind, err := urlMappingQuery.UrlToKindCache(appId, spanAttrs.HttpRoute())
	// if err != nil {
	// 	log.Errorf("获取url kind失败: %s %s", appAlias, err)
	// 	return
	// }
	t, err := urlMappingQuery.UrlToTypeFast(appId, spanAttrs.HttpRoute())
	if err != nil {
		log.Errorf("获取url type失败: %s %s", appAlias, err)
		return
	}
	// fmt.Println(appId, appAlias, serviceName.AsString(), kind, t, spanAttrs.HttpRoute(), spanAttrs.StatusCode(), span.Status().Code().String())
	arr := strings.Split(spanAttrs.HttpRoute(), "/")
	h.Float64Histogram.Record(context.Background(), float64(duration), metric.WithAttributes(
		attribute.Int64("app.id", appId),
		attribute.String("app.alias", appAlias),
		attribute.String("service.name", serviceName.AsString()),
		// attribute.String("url.kind", kind),
		attribute.Int("url.type", t),
		attribute.String("url.method", spanAttrs.HttpMethod()),
		attribute.String("url.prefix", spanAttrs.HttpRoute()),
		attribute.Int("url.level", len(arr)-1),
		attribute.Int("url.is_perfect_match", 1),
		attribute.Int("status_code", int(spanAttrs.StatusCode())),
		attribute.Bool("error", spanAttrs.StatusCode() >= 400 || span.Status().Code().String() == "Error"),
	))
	for len(arr) > 1 {
		arr = arr[:len(arr)-1]
		subroute := strings.Join(arr, "/")
		if subroute == "" {
			subroute = "/"
		}
		h.Float64Histogram.Record(context.Background(), float64(duration), metric.WithAttributes(
			attribute.Int64("app.id", appId),
			attribute.String("app.alias", appAlias),
			attribute.String("service.name", serviceName.AsString()),
			// attribute.String("url.kind", kind),
			attribute.Int("url.type", t),
			attribute.String("url.method", spanAttrs.HttpMethod()),
			attribute.String("url.prefix", subroute),
			attribute.Int("url.level", len(arr)-1),
			attribute.Int("url.is_perfect_match", 0),
			attribute.Int("status_code", int(spanAttrs.StatusCode())),
			attribute.Bool("error", spanAttrs.StatusCode() >= 400 || span.Status().Code().String() == "Error"),
		))

	}
}