|
|
@@ -3,12 +3,12 @@ package main
|
|
|
import (
|
|
|
"context"
|
|
|
"encoding/hex"
|
|
|
+ mpb "git.cestong.com.cn/cecf/trace-stream-creator/pkg/pb"
|
|
|
"github.com/ClickHouse/clickhouse-go/v2"
|
|
|
+ "github.com/pkg/errors"
|
|
|
"github.com/sirupsen/logrus"
|
|
|
pcommon "go.opentelemetry.io/proto/otlp/common/v1"
|
|
|
presource "go.opentelemetry.io/proto/otlp/resource/v1"
|
|
|
-
|
|
|
- mpb "git.cestong.com.cn/cecf/trace-stream-creator/pkg/pb"
|
|
|
ptrace "go.opentelemetry.io/proto/otlp/trace/v1"
|
|
|
)
|
|
|
|
|
|
@@ -26,18 +26,35 @@ func NewTraceLoader(chCli clickhouse.Conn, traceIdChan chan timeRangedTraceID, f
|
|
|
}
|
|
|
|
|
|
func (tl *TraceLoader) start(parentCtx context.Context) {
|
|
|
- for range tl.parallelism {
|
|
|
+ for i := 0; i < tl.parallelism; i++ {
|
|
|
go tl.startHandleTraceID(parentCtx)
|
|
|
}
|
|
|
tl.logger.Infof("started, parallelism: %d", tl.parallelism)
|
|
|
}
|
|
|
|
|
|
-func (tl *TraceLoader) loadTrace(traceID timeRangedTraceID) (*mpb.FlatTrace, error) {
|
|
|
- tl.logger.Debugf("load trace %s", traceID.TraceID)
|
|
|
- sql := ""
|
|
|
- rows, errQuerySpans := tl.chCli.Query(context.Background(), sql, traceID)
|
|
|
- var tb TraceBuilder
|
|
|
- ft := tb.buildTrace()
|
|
|
+func (tl *TraceLoader) loadTrace(rangedTraceID timeRangedTraceID) (*mpb.FlatTrace, error) {
|
|
|
+ tl.logger.Debugf("load trace %s", rangedTraceID.TraceID)
|
|
|
+ sql := " select Timestamp, TraceId, SpanId, ParentSpanId, TraceState, SpanName, SpanKind, ServiceName, " +
|
|
|
+ "ResourceAttributes, ScopeName, ScopeVersion, SpanAttributes, Duration, StatusCode, StatusMessage, " +
|
|
|
+ "`Events.Timestamp`, `Events.Name`, `Events.Attributes`, `Links.TraceId`, `Links.SpanId`, `Links.TraceState`, " +
|
|
|
+ "`Links.Attributes`, HttpCode, HttpMethod, HttpURL, ContainerId, srcIP, srcPort, targetIP, targetPort, " +
|
|
|
+ "RPCType, RPCName, RPCRequest, RPCResult, FuncNameSpace, FuncName, FuncLineNO, FuncResult, dbStatement, " +
|
|
|
+ "dbConnectionString, `Exceptions.type`, `Exceptions.message`, `Exceptions.stacktrace`, AppAlias " +
|
|
|
+ "from otel.traces where Timestamp < ? and Timestamp > ? and TraceId=?"
|
|
|
+
|
|
|
+ rows, errQuerySpans := tl.chCli.Query(context.Background(), sql, rangedTraceID.end, rangedTraceID.begin, rangedTraceID.TraceID)
|
|
|
+ if errQuerySpans != nil {
|
|
|
+ return nil, errors.WithMessagef(errQuerySpans, "query sql %s", sql)
|
|
|
+ }
|
|
|
+ chSpans := make([]ChSpan, 0, 100)
|
|
|
+ for rows.Next() {
|
|
|
+ var span ChSpan
|
|
|
+ if errScan := rows.ScanStruct(&span); errScan != nil {
|
|
|
+ return nil, errors.WithMessagef(errScan, "scan to ChSpan")
|
|
|
+ }
|
|
|
+ chSpans = append(chSpans, span)
|
|
|
+ }
|
|
|
+ ft := tl.buildTrace(rangedTraceID.TraceID, chSpans)
|
|
|
return ft, nil
|
|
|
}
|
|
|
|
|
|
@@ -66,45 +83,137 @@ func (tl *TraceLoader) handleFlatTrace(ft *mpb.FlatTrace) error {
|
|
|
return nil
|
|
|
}
|
|
|
|
|
|
-type TraceBuilder struct {
|
|
|
-}
|
|
|
-
|
|
|
-func (tb *TraceBuilder) buildTrace() *mpb.FlatTrace {
|
|
|
- span := ptrace.Span{
|
|
|
- TraceId: nil,
|
|
|
- SpanId: nil,
|
|
|
- TraceState: "",
|
|
|
- ParentSpanId: nil,
|
|
|
- Flags: 0,
|
|
|
- Name: "",
|
|
|
- Kind: 0,
|
|
|
- StartTimeUnixNano: 0,
|
|
|
- EndTimeUnixNano: 0,
|
|
|
- Attributes: nil,
|
|
|
- DroppedAttributesCount: 0,
|
|
|
- Events: nil,
|
|
|
- DroppedEventsCount: 0,
|
|
|
- Links: nil,
|
|
|
- DroppedLinksCount: 0,
|
|
|
- Status: nil,
|
|
|
- }
|
|
|
- scope := pcommon.InstrumentationScope{
|
|
|
- Name: "scopeName",
|
|
|
- Version: "scopeVersion",
|
|
|
- DroppedAttributesCount: 0,
|
|
|
- Attributes: make([]*pcommon.KeyValue, 0),
|
|
|
- }
|
|
|
- resource := presource.Resource{Attributes: make([]*pcommon.KeyValue, 0)}
|
|
|
- flatSpan := mpb.FlatSpan{
|
|
|
- Span: &span,
|
|
|
- Scope: &scope,
|
|
|
- Resource: &resource,
|
|
|
+func (tl *TraceLoader) buildTrace(traceID string, chSpans []ChSpan) *mpb.FlatTrace {
|
|
|
+ flatSpans := make([]*mpb.FlatSpan, 0, len(chSpans))
|
|
|
+ for _, chSpan := range chSpans {
|
|
|
+ attr := make([]*pcommon.KeyValue, 0, len(chSpan.SpanAttributes))
|
|
|
+ for sa, saValue := range chSpan.SpanAttributes {
|
|
|
+ attr = append(attr, &pcommon.KeyValue{
|
|
|
+ Key: sa,
|
|
|
+ Value: &pcommon.AnyValue{
|
|
|
+ Value: &pcommon.AnyValue_StringValue{StringValue: saValue},
|
|
|
+ },
|
|
|
+ })
|
|
|
+ }
|
|
|
+ evs := make([]*ptrace.Span_Event, 0, len(chSpan.EventsTimestamp))
|
|
|
+ if !(len(chSpan.EventsTimestamp) == len(chSpan.EventsName) &&
|
|
|
+ len(chSpan.EventsTimestamp) == len(chSpan.EventsAttributes)) {
|
|
|
+ tl.logger.Errorf("events.name,timestamp,attributes长度不相同")
|
|
|
+ continue
|
|
|
+ }
|
|
|
+ for i, et := range chSpan.EventsTimestamp {
|
|
|
+ evName := chSpan.EventsName[i]
|
|
|
+ evAttr := chSpan.EventsAttributes[i]
|
|
|
+ spanEvAttrs := make([]*pcommon.KeyValue, 0, len(evAttr))
|
|
|
+ for evAttrKey, evAttrValue := range evAttr {
|
|
|
+ spanEvAttrs = append(spanEvAttrs, &pcommon.KeyValue{
|
|
|
+ Key: evAttrKey,
|
|
|
+ Value: &pcommon.AnyValue{Value: &pcommon.AnyValue_StringValue{StringValue: evAttrValue}},
|
|
|
+ })
|
|
|
+ }
|
|
|
+ evs = append(evs, &ptrace.Span_Event{
|
|
|
+ TimeUnixNano: uint64(et.UnixNano()),
|
|
|
+ Name: evName,
|
|
|
+ Attributes: spanEvAttrs,
|
|
|
+ DroppedAttributesCount: 0,
|
|
|
+ })
|
|
|
+ }
|
|
|
+ linkSize := len(chSpan.LinksSpanId)
|
|
|
+ if !(linkSize == len(chSpan.LinksTraceId) &&
|
|
|
+ linkSize == len(chSpan.LinksTraceState) &&
|
|
|
+ linkSize == len(chSpan.LinksAttributes)) {
|
|
|
+ tl.logger.Errorf("link.spanId,traceId,attributes,state长度不相同")
|
|
|
+ continue
|
|
|
+ }
|
|
|
+ lks := make([]*ptrace.Span_Link, 0, linkSize)
|
|
|
+ for i, linkSpanId := range chSpan.LinksSpanId {
|
|
|
+ linkAttr := make([]*pcommon.KeyValue, 0, linkSize)
|
|
|
+ for linkKey, linkValue := range chSpan.LinksAttributes[i] {
|
|
|
+ linkAttr = append(linkAttr, &pcommon.KeyValue{
|
|
|
+ Key: linkKey,
|
|
|
+ Value: &pcommon.AnyValue{Value: &pcommon.AnyValue_StringValue{StringValue: linkValue}},
|
|
|
+ })
|
|
|
+ }
|
|
|
+ linkTraceID, errLinkTraceID := hexIDStr2Bytes(chSpan.LinksTraceId[i])
|
|
|
+ linkSpanID, errLinkSpanID := hexIDStr2Bytes(linkSpanId)
|
|
|
+ if errLinkTraceID != nil || errLinkSpanID != nil {
|
|
|
+ tl.logger.Errorf("link has invalid spanID/traceID, err:%v-%v", errLinkSpanID, errLinkSpanID)
|
|
|
+ continue
|
|
|
+ }
|
|
|
+ lks = append(lks, &ptrace.Span_Link{
|
|
|
+ TraceId: linkTraceID,
|
|
|
+ SpanId: linkSpanID,
|
|
|
+ TraceState: chSpan.LinksTraceState[i],
|
|
|
+ Attributes: linkAttr,
|
|
|
+ DroppedAttributesCount: 0,
|
|
|
+ Flags: 0,
|
|
|
+ })
|
|
|
+ }
|
|
|
+ spanTraceID, errSpanTraceID := hexIDStr2Bytes(chSpan.TraceId)
|
|
|
+ spanSpanID, errSpanSpanID := hexIDStr2Bytes(chSpan.SpanId)
|
|
|
+ spanParentSpanID, errSpanParentSpanID := hexIDStr2Bytes(chSpan.ParentSpanId)
|
|
|
+ if errSpanTraceID != nil || errSpanSpanID != nil || errSpanParentSpanID != nil {
|
|
|
+ tl.logger.Errorf("span has invalid tid/sid/psid:%v-%v-%v", errSpanTraceID, errSpanSpanID, errSpanParentSpanID)
|
|
|
+ continue
|
|
|
+ }
|
|
|
+ span := ptrace.Span{
|
|
|
+ TraceId: spanTraceID,
|
|
|
+ SpanId: spanSpanID,
|
|
|
+ TraceState: "",
|
|
|
+ ParentSpanId: spanParentSpanID,
|
|
|
+ Flags: 0,
|
|
|
+ Name: chSpan.SpanName,
|
|
|
+ Kind: ptrace.Span_SpanKind(chSpan.SpanKind),
|
|
|
+ StartTimeUnixNano: uint64(chSpan.Timestamp.UnixNano()),
|
|
|
+ EndTimeUnixNano: uint64(chSpan.Timestamp.UnixNano() + chSpan.Duration),
|
|
|
+ Attributes: attr,
|
|
|
+ DroppedAttributesCount: 0,
|
|
|
+ Events: evs,
|
|
|
+ DroppedEventsCount: 0,
|
|
|
+ Links: lks,
|
|
|
+ DroppedLinksCount: 0,
|
|
|
+ Status: &ptrace.Status{
|
|
|
+ Message: chSpan.StatusMessage,
|
|
|
+ Code: ptrace.Status_StatusCode(chSpan.StatusCode),
|
|
|
+ },
|
|
|
+ }
|
|
|
+ scope := pcommon.InstrumentationScope{
|
|
|
+ Name: chSpan.ScopeName,
|
|
|
+ Version: chSpan.ScopeVersion,
|
|
|
+ DroppedAttributesCount: 0,
|
|
|
+ Attributes: make([]*pcommon.KeyValue, 0),
|
|
|
+ }
|
|
|
+ resource := presource.Resource{Attributes: make([]*pcommon.KeyValue, 0, len(chSpan.ResourceAttributes)), DroppedAttributesCount: 0}
|
|
|
+ for raKey, raValue := range chSpan.ResourceAttributes {
|
|
|
+ resource.Attributes = append(resource.Attributes, &pcommon.KeyValue{
|
|
|
+ Key: raKey,
|
|
|
+ Value: &pcommon.AnyValue{
|
|
|
+ Value: &pcommon.AnyValue_StringValue{StringValue: raValue},
|
|
|
+ },
|
|
|
+ })
|
|
|
+ }
|
|
|
+ flatSpan := mpb.FlatSpan{
|
|
|
+ Span: &span,
|
|
|
+ Scope: &scope,
|
|
|
+ Resource: &resource,
|
|
|
+ }
|
|
|
+ flatSpans = append(flatSpans, &flatSpan)
|
|
|
}
|
|
|
- traceID := "traceID"
|
|
|
- traceIDBytes, _ := hex.DecodeString(traceID)
|
|
|
+ traceIDBytes, _ := hexIDStr2Bytes(traceID)
|
|
|
tr := mpb.FlatTrace{
|
|
|
TraceId: traceIDBytes,
|
|
|
- FlatSpans: []*mpb.FlatSpan{&flatSpan},
|
|
|
+ FlatSpans: flatSpans,
|
|
|
}
|
|
|
return &tr
|
|
|
}
|
|
|
+
|
|
|
+func hexIDStr2Bytes(idStr string) ([]byte, error) {
|
|
|
+ if idStr == "" {
|
|
|
+ return nil, nil
|
|
|
+ }
|
|
|
+ bs, err := hex.DecodeString(idStr)
|
|
|
+ if err != nil {
|
|
|
+ return nil, errors.WithMessagef(err, "hex decode:[%s]", idStr)
|
|
|
+ }
|
|
|
+ return bs, nil
|
|
|
+}
|
liubing