model.go 8.5 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311
  1. // Copyright The OpenTelemetry Authors
  2. //
  3. // Licensed under the Apache License, Version 2.0 (the "License");
  4. // you may not use this file except in compliance with the License.
  5. // You may obtain a copy of the License at
  6. //
  7. // http://www.apache.org/licenses/LICENSE-2.0
  8. //
  9. // Unless required by applicable law or agreed to in writing, software
  10. // distributed under the License is distributed on an "AS IS" BASIS,
  11. // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12. // See the License for the specific language governing permissions and
  13. // limitations under the License.
  14. package zipkin // import "go.opentelemetry.io/otel/exporters/zipkin"
  15. import (
  16. "encoding/binary"
  17. "encoding/json"
  18. "fmt"
  19. "net"
  20. "strconv"
  21. "strings"
  22. zkmodel "github.com/openzipkin/zipkin-go/model"
  23. "go.opentelemetry.io/otel/attribute"
  24. "go.opentelemetry.io/otel/codes"
  25. "go.opentelemetry.io/otel/sdk/resource"
  26. tracesdk "go.opentelemetry.io/otel/sdk/trace"
  27. semconv "go.opentelemetry.io/otel/semconv/v1.21.0"
  28. "go.opentelemetry.io/otel/trace"
  29. )
  30. const (
  31. keyInstrumentationLibraryName = "otel.library.name"
  32. keyInstrumentationLibraryVersion = "otel.library.version"
  33. keyPeerHostname attribute.Key = "peer.hostname"
  34. keyPeerAddress attribute.Key = "peer.address"
  35. )
  36. var defaultServiceName string
  37. func init() {
  38. // fetch service.name from default resource for backup
  39. defaultResource := resource.Default()
  40. if value, exists := defaultResource.Set().Value(semconv.ServiceNameKey); exists {
  41. defaultServiceName = value.AsString()
  42. }
  43. }
  44. // SpanModels converts OpenTelemetry spans into Zipkin model spans.
  45. // This is used for exporting to Zipkin compatible tracing services.
  46. func SpanModels(batch []tracesdk.ReadOnlySpan) []zkmodel.SpanModel {
  47. models := make([]zkmodel.SpanModel, 0, len(batch))
  48. for _, data := range batch {
  49. models = append(models, toZipkinSpanModel(data))
  50. }
  51. return models
  52. }
  53. func getServiceName(attrs []attribute.KeyValue) string {
  54. for _, kv := range attrs {
  55. if kv.Key == semconv.ServiceNameKey {
  56. return kv.Value.AsString()
  57. }
  58. }
  59. return defaultServiceName
  60. }
  61. func toZipkinSpanModel(data tracesdk.ReadOnlySpan) zkmodel.SpanModel {
  62. return zkmodel.SpanModel{
  63. SpanContext: toZipkinSpanContext(data),
  64. Name: data.Name(),
  65. Kind: toZipkinKind(data.SpanKind()),
  66. Timestamp: data.StartTime(),
  67. Duration: data.EndTime().Sub(data.StartTime()),
  68. Shared: false,
  69. LocalEndpoint: &zkmodel.Endpoint{
  70. ServiceName: getServiceName(data.Resource().Attributes()),
  71. },
  72. RemoteEndpoint: toZipkinRemoteEndpoint(data),
  73. Annotations: toZipkinAnnotations(data.Events()),
  74. Tags: toZipkinTags(data),
  75. }
  76. }
  77. func toZipkinSpanContext(data tracesdk.ReadOnlySpan) zkmodel.SpanContext {
  78. return zkmodel.SpanContext{
  79. TraceID: toZipkinTraceID(data.SpanContext().TraceID()),
  80. ID: toZipkinID(data.SpanContext().SpanID()),
  81. ParentID: toZipkinParentID(data.Parent().SpanID()),
  82. Debug: false,
  83. Sampled: nil,
  84. Err: nil,
  85. }
  86. }
  87. func toZipkinTraceID(traceID trace.TraceID) zkmodel.TraceID {
  88. return zkmodel.TraceID{
  89. High: binary.BigEndian.Uint64(traceID[:8]),
  90. Low: binary.BigEndian.Uint64(traceID[8:]),
  91. }
  92. }
  93. func toZipkinID(spanID trace.SpanID) zkmodel.ID {
  94. return zkmodel.ID(binary.BigEndian.Uint64(spanID[:]))
  95. }
  96. func toZipkinParentID(spanID trace.SpanID) *zkmodel.ID {
  97. if spanID.IsValid() {
  98. id := toZipkinID(spanID)
  99. return &id
  100. }
  101. return nil
  102. }
  103. func toZipkinKind(kind trace.SpanKind) zkmodel.Kind {
  104. switch kind {
  105. case trace.SpanKindUnspecified:
  106. return zkmodel.Undetermined
  107. case trace.SpanKindInternal:
  108. // The spec says we should set the kind to nil, but
  109. // the model does not allow that.
  110. return zkmodel.Undetermined
  111. case trace.SpanKindServer:
  112. return zkmodel.Server
  113. case trace.SpanKindClient:
  114. return zkmodel.Client
  115. case trace.SpanKindProducer:
  116. return zkmodel.Producer
  117. case trace.SpanKindConsumer:
  118. return zkmodel.Consumer
  119. }
  120. return zkmodel.Undetermined
  121. }
  122. func toZipkinAnnotations(events []tracesdk.Event) []zkmodel.Annotation {
  123. if len(events) == 0 {
  124. return nil
  125. }
  126. annotations := make([]zkmodel.Annotation, 0, len(events))
  127. for _, event := range events {
  128. value := event.Name
  129. if len(event.Attributes) > 0 {
  130. jsonString := attributesToJSONMapString(event.Attributes)
  131. if jsonString != "" {
  132. value = fmt.Sprintf("%s: %s", event.Name, jsonString)
  133. }
  134. }
  135. annotations = append(annotations, zkmodel.Annotation{
  136. Timestamp: event.Time,
  137. Value: value,
  138. })
  139. }
  140. return annotations
  141. }
  142. func attributesToJSONMapString(attributes []attribute.KeyValue) string {
  143. m := make(map[string]interface{}, len(attributes))
  144. for _, a := range attributes {
  145. m[(string)(a.Key)] = a.Value.AsInterface()
  146. }
  147. // if an error happens, the result will be an empty string
  148. jsonBytes, _ := json.Marshal(m)
  149. return (string)(jsonBytes)
  150. }
  151. // attributeToStringPair serializes each attribute to a string pair.
  152. func attributeToStringPair(kv attribute.KeyValue) (string, string) {
  153. switch kv.Value.Type() {
  154. // For slice attributes, serialize as JSON list string.
  155. case attribute.BOOLSLICE:
  156. data, _ := json.Marshal(kv.Value.AsBoolSlice())
  157. return (string)(kv.Key), (string)(data)
  158. case attribute.INT64SLICE:
  159. data, _ := json.Marshal(kv.Value.AsInt64Slice())
  160. return (string)(kv.Key), (string)(data)
  161. case attribute.FLOAT64SLICE:
  162. data, _ := json.Marshal(kv.Value.AsFloat64Slice())
  163. return (string)(kv.Key), (string)(data)
  164. case attribute.STRINGSLICE:
  165. data, _ := json.Marshal(kv.Value.AsStringSlice())
  166. return (string)(kv.Key), (string)(data)
  167. default:
  168. return (string)(kv.Key), kv.Value.Emit()
  169. }
  170. }
  171. // extraZipkinTags are those that may be added to every outgoing span.
  172. var extraZipkinTags = []string{
  173. "otel.status_code",
  174. keyInstrumentationLibraryName,
  175. keyInstrumentationLibraryVersion,
  176. }
  177. func toZipkinTags(data tracesdk.ReadOnlySpan) map[string]string {
  178. attr := data.Attributes()
  179. resourceAttr := data.Resource().Attributes()
  180. m := make(map[string]string, len(attr)+len(resourceAttr)+len(extraZipkinTags))
  181. for _, kv := range attr {
  182. k, v := attributeToStringPair(kv)
  183. m[k] = v
  184. }
  185. for _, kv := range resourceAttr {
  186. k, v := attributeToStringPair(kv)
  187. m[k] = v
  188. }
  189. if data.Status().Code != codes.Unset {
  190. // Zipkin expect to receive uppercase status values
  191. // rather than default capitalized ones.
  192. m["otel.status_code"] = strings.ToUpper(data.Status().Code.String())
  193. }
  194. if data.Status().Code == codes.Error {
  195. m["error"] = data.Status().Description
  196. } else {
  197. delete(m, "error")
  198. }
  199. if is := data.InstrumentationScope(); is.Name != "" {
  200. m[keyInstrumentationLibraryName] = is.Name
  201. if is.Version != "" {
  202. m[keyInstrumentationLibraryVersion] = is.Version
  203. }
  204. }
  205. if len(m) == 0 {
  206. return nil
  207. }
  208. return m
  209. }
  210. // Rank determines selection order for remote endpoint. See the specification
  211. // https://github.com/open-telemetry/opentelemetry-specification/blob/v1.20.0/specification/trace/sdk_exporters/zipkin.md#otlp---zipkin
  212. var remoteEndpointKeyRank = map[attribute.Key]int{
  213. semconv.PeerServiceKey: 0,
  214. semconv.NetPeerNameKey: 1,
  215. semconv.NetSockPeerNameKey: 2,
  216. semconv.NetSockPeerAddrKey: 3,
  217. keyPeerHostname: 4,
  218. keyPeerAddress: 5,
  219. semconv.DBNameKey: 6,
  220. }
  221. func toZipkinRemoteEndpoint(data tracesdk.ReadOnlySpan) *zkmodel.Endpoint {
  222. // Should be set only for client or producer kind
  223. if sk := data.SpanKind(); sk != trace.SpanKindClient && sk != trace.SpanKindProducer {
  224. return nil
  225. }
  226. attr := data.Attributes()
  227. var endpointAttr attribute.KeyValue
  228. for _, kv := range attr {
  229. rank, ok := remoteEndpointKeyRank[kv.Key]
  230. if !ok {
  231. continue
  232. }
  233. currentKeyRank, ok := remoteEndpointKeyRank[endpointAttr.Key]
  234. if ok && rank < currentKeyRank {
  235. endpointAttr = kv
  236. } else if !ok {
  237. endpointAttr = kv
  238. }
  239. }
  240. if endpointAttr.Key == "" {
  241. return nil
  242. }
  243. if endpointAttr.Key != semconv.NetSockPeerAddrKey &&
  244. endpointAttr.Value.Type() == attribute.STRING {
  245. return &zkmodel.Endpoint{
  246. ServiceName: endpointAttr.Value.AsString(),
  247. }
  248. }
  249. return remoteEndpointPeerIPWithPort(endpointAttr.Value.AsString(), attr)
  250. }
  251. // Handles `net.peer.ip` remote endpoint separately (should include `net.peer.ip`
  252. // as well, if available).
  253. func remoteEndpointPeerIPWithPort(peerIP string, attrs []attribute.KeyValue) *zkmodel.Endpoint {
  254. ip := net.ParseIP(peerIP)
  255. if ip == nil {
  256. return nil
  257. }
  258. endpoint := &zkmodel.Endpoint{}
  259. // Determine if IPv4 or IPv6
  260. if ip.To4() != nil {
  261. endpoint.IPv4 = ip
  262. } else {
  263. endpoint.IPv6 = ip
  264. }
  265. for _, kv := range attrs {
  266. if kv.Key == semconv.NetSockPeerPortKey {
  267. port, _ := strconv.ParseUint(kv.Value.Emit(), 10, 16)
  268. endpoint.Port = uint16(port)
  269. return endpoint
  270. }
  271. }
  272. return endpoint
  273. }