// Copyright 2019 The Go Authors. All rights reserved. // Use of this source code is governed by a BSD-style // license that can be found in the LICENSE file. // Package ocagent adds the ability to export all telemetry to an ocagent. // This keeps the complie time dependencies to zero and allows the agent to // have the exporters needed for telemetry aggregation and viewing systems. package ocagent import ( "bytes" "encoding/json" "fmt" "net/http" "os" "time" "golang.org/x/tools/internal/lsp/telemetry/log" "golang.org/x/tools/internal/lsp/telemetry/metric" "golang.org/x/tools/internal/lsp/telemetry/ocagent/wire" "golang.org/x/tools/internal/lsp/telemetry/tag" "golang.org/x/tools/internal/lsp/telemetry/trace" "golang.org/x/tools/internal/lsp/telemetry/worker" ) const DefaultAddress = "http://localhost:55678" const exportRate = 2 * time.Second type exporter struct { address string node *wire.Node spans []*wire.Span metrics []*wire.Metric } func Export(service, address string) { if address == "off" { return } hostname, _ := os.Hostname() exporter := &exporter{ address: address, node: &wire.Node{ Identifier: &wire.ProcessIdentifier{ HostName: hostname, Pid: uint32(os.Getpid()), StartTimestamp: convertTimestamp(time.Now()), }, LibraryInfo: &wire.LibraryInfo{ Language: wire.LanguageGo, ExporterVersion: "0.0.1", CoreLibraryVersion: "x/tools", }, ServiceInfo: &wire.ServiceInfo{ Name: service, }, }, } if exporter.address == "" { exporter.address = DefaultAddress } //TODO: add metrics once the ocagent json metric interface works trace.RegisterObservers(exporter.observeTrace) go func() { for _ = range time.Tick(exportRate) { worker.Do(func() { exporter.flush() }) } }() } func (e *exporter) observeTrace(span *trace.Span) { // is this a completed span? if span.Finish.IsZero() { return } e.spans = append(e.spans, convertSpan(span)) } func (e *exporter) observeMetric(data metric.Data) { e.metrics = append(e.metrics, convertMetric(data)) } func (e *exporter) flush() { spans := e.spans e.spans = nil metrics := e.metrics e.metrics = nil if len(spans) > 0 { e.send("/v1/trace", &wire.ExportTraceServiceRequest{ Node: e.node, Spans: spans, //TODO: Resource? }) } if len(metrics) > 0 { e.send("/v1/metrics", &wire.ExportMetricsServiceRequest{ Node: e.node, Metrics: metrics, //TODO: Resource? }) } } func (e *exporter) send(endpoint string, message interface{}) { blob, err := json.Marshal(message) if err != nil { errorInExport("ocagent failed to marshal message for %v: %v", endpoint, err) return } uri := e.address + endpoint req, err := http.NewRequest("POST", uri, bytes.NewReader(blob)) if err != nil { errorInExport("ocagent failed to build request for %v: %v", uri, err) return } req.Header.Set("Content-Type", "application/json") res, err := http.DefaultClient.Do(req) if err != nil { errorInExport("ocagent failed to send message: %v \n", err) return } res.Body.Close() return } func errorInExport(message string, args ...interface{}) { // This function is useful when debugging the exporter, but in general we // want to just drop any export } func convertTimestamp(t time.Time) wire.Timestamp { return t.Format(time.RFC3339Nano) } func toTruncatableString(s string) *wire.TruncatableString { if s == "" { return nil } return &wire.TruncatableString{Value: s} } func convertSpan(span *trace.Span) *wire.Span { result := &wire.Span{ TraceId: span.TraceID[:], SpanId: span.SpanID[:], TraceState: nil, //TODO? ParentSpanId: span.ParentID[:], Name: toTruncatableString(span.Name), Kind: wire.UnspecifiedSpanKind, StartTime: convertTimestamp(span.Start), EndTime: convertTimestamp(span.Finish), Attributes: convertAttributes(span.Tags), TimeEvents: convertEvents(span.Events), SameProcessAsParentSpan: true, //TODO: StackTrace? //TODO: Links? //TODO: Status? //TODO: Resource? } return result } func convertMetric(data metric.Data) *wire.Metric { return nil //TODO: } func convertAttributes(tags tag.List) *wire.Attributes { if len(tags) == 0 { return nil } attributes := make(map[string]wire.Attribute) for _, tag := range tags { attributes[fmt.Sprint(tag.Key)] = convertAttribute(tag.Value) } return &wire.Attributes{AttributeMap: attributes} } func convertAttribute(v interface{}) wire.Attribute { switch v := v.(type) { case int8: return wire.IntAttribute{IntValue: int64(v)} case int16: return wire.IntAttribute{IntValue: int64(v)} case int32: return wire.IntAttribute{IntValue: int64(v)} case int64: return wire.IntAttribute{IntValue: v} case int: return wire.IntAttribute{IntValue: int64(v)} case uint8: return wire.IntAttribute{IntValue: int64(v)} case uint16: return wire.IntAttribute{IntValue: int64(v)} case uint32: return wire.IntAttribute{IntValue: int64(v)} case uint64: return wire.IntAttribute{IntValue: int64(v)} case uint: return wire.IntAttribute{IntValue: int64(v)} case float32: return wire.DoubleAttribute{DoubleValue: float64(v)} case float64: return wire.DoubleAttribute{DoubleValue: v} case bool: return wire.BoolAttribute{BoolValue: v} case string: return wire.StringAttribute{StringValue: toTruncatableString(v)} default: return wire.StringAttribute{StringValue: toTruncatableString(fmt.Sprint(v))} } } func convertEvents(events []trace.Event) *wire.TimeEvents { //TODO: MessageEvents? result := make([]wire.TimeEvent, len(events)) for i, event := range events { result[i] = convertEvent(event) } return &wire.TimeEvents{TimeEvent: result} } func convertEvent(event trace.Event) wire.TimeEvent { return wire.TimeEvent{ Time: convertTimestamp(event.Time), Annotation: convertAnnotation(event.Tags), } } func convertAnnotation(tags tag.List) *wire.Annotation { if len(tags) == 0 { return nil } entry := log.ToEntry(nil, time.Time{}, tags) description := entry.Message if description == "" && entry.Error != nil { description = entry.Error.Error() entry.Error = nil } tags = entry.Tags if entry.Error != nil { tags = append(tags, tag.Of("Error", entry.Error)) } return &wire.Annotation{ Description: toTruncatableString(description), Attributes: convertAttributes(tags), } }