// 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 compile time dependencies to zero and allows the agent to // have the exporters needed for telemetry aggregation and viewing systems. package ocagent import ( "bytes" "context" "encoding/json" "fmt" "net/http" "os" "path/filepath" "sync" "time" "golang.org/x/tools/internal/event" "golang.org/x/tools/internal/event/core" "golang.org/x/tools/internal/event/export" "golang.org/x/tools/internal/event/export/metric" "golang.org/x/tools/internal/event/export/ocagent/wire" "golang.org/x/tools/internal/event/keys" "golang.org/x/tools/internal/event/label" ) type Config struct { Start time.Time Host string Process uint32 Client *http.Client Service string Address string Rate time.Duration } var ( connectMu sync.Mutex exporters = make(map[Config]*Exporter) ) // Discover finds the local agent to export to, it will return nil if there // is not one running. // TODO: Actually implement a discovery protocol rather than a hard coded address func Discover() *Config { return &Config{ Address: "http://localhost:55678", } } type Exporter struct { mu sync.Mutex config Config spans []*export.Span metrics []metric.Data } // Connect creates a process specific exporter with the specified // serviceName and the address of the ocagent to which it will upload // its telemetry. func Connect(config *Config) *Exporter { if config == nil || config.Address == "off" { return nil } resolved := *config if resolved.Host == "" { hostname, _ := os.Hostname() resolved.Host = hostname } if resolved.Process == 0 { resolved.Process = uint32(os.Getpid()) } if resolved.Client == nil { resolved.Client = http.DefaultClient } if resolved.Service == "" { resolved.Service = filepath.Base(os.Args[0]) } if resolved.Rate == 0 { resolved.Rate = 2 * time.Second } connectMu.Lock() defer connectMu.Unlock() if exporter, found := exporters[resolved]; found { return exporter } exporter := &Exporter{config: resolved} exporters[resolved] = exporter if exporter.config.Start.IsZero() { exporter.config.Start = time.Now() } go func() { for range time.Tick(exporter.config.Rate) { exporter.Flush() } }() return exporter } func (e *Exporter) ProcessEvent(ctx context.Context, ev core.Event, lm label.Map) context.Context { switch { case event.IsEnd(ev): e.mu.Lock() defer e.mu.Unlock() span := export.GetSpan(ctx) if span != nil { e.spans = append(e.spans, span) } case event.IsMetric(ev): e.mu.Lock() defer e.mu.Unlock() data := metric.Entries.Get(lm).([]metric.Data) e.metrics = append(e.metrics, data...) } return ctx } func (e *Exporter) Flush() { e.mu.Lock() defer e.mu.Unlock() spans := make([]*wire.Span, len(e.spans)) for i, s := range e.spans { spans[i] = convertSpan(s) } e.spans = nil metrics := make([]*wire.Metric, len(e.metrics)) for i, m := range e.metrics { metrics[i] = convertMetric(m, e.config.Start) } e.metrics = nil if len(spans) > 0 { e.send("/v1/trace", &wire.ExportTraceServiceRequest{ Node: e.config.buildNode(), Spans: spans, //TODO: Resource? }) } if len(metrics) > 0 { e.send("/v1/metrics", &wire.ExportMetricsServiceRequest{ Node: e.config.buildNode(), Metrics: metrics, //TODO: Resource? }) } } func (cfg *Config) buildNode() *wire.Node { return &wire.Node{ Identifier: &wire.ProcessIdentifier{ HostName: cfg.Host, Pid: cfg.Process, StartTimestamp: convertTimestamp(cfg.Start), }, LibraryInfo: &wire.LibraryInfo{ Language: wire.LanguageGo, ExporterVersion: "0.0.1", CoreLibraryVersion: "x/tools", }, ServiceInfo: &wire.ServiceInfo{ Name: cfg.Service, }, } } 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.config.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 := e.config.Client.Do(req) if err != nil { errorInExport("ocagent failed to send message: %v \n", err) return } if res.Body != nil { res.Body.Close() } } 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 *export.Span) *wire.Span { result := &wire.Span{ TraceID: span.ID.TraceID[:], SpanID: span.ID.SpanID[:], TraceState: nil, //TODO? ParentSpanID: span.ParentID[:], Name: toTruncatableString(span.Name), Kind: wire.UnspecifiedSpanKind, StartTime: convertTimestamp(span.Start().At()), EndTime: convertTimestamp(span.Finish().At()), Attributes: convertAttributes(span.Start(), 1), TimeEvents: convertEvents(span.Events()), SameProcessAsParentSpan: true, //TODO: StackTrace? //TODO: Links? //TODO: Status? //TODO: Resource? } return result } func convertMetric(data metric.Data, start time.Time) *wire.Metric { descriptor := dataToMetricDescriptor(data) timeseries := dataToTimeseries(data, start) if descriptor == nil && timeseries == nil { return nil } // TODO: handle Histogram metrics return &wire.Metric{ MetricDescriptor: descriptor, Timeseries: timeseries, // TODO: attach Resource? } } func skipToValidLabel(list label.List, index int) (int, label.Label) { // skip to the first valid label for ; list.Valid(index); index++ { l := list.Label(index) if !l.Valid() || l.Key() == keys.Label { continue } return index, l } return -1, label.Label{} } func convertAttributes(list label.List, index int) *wire.Attributes { index, l := skipToValidLabel(list, index) if !l.Valid() { return nil } attributes := make(map[string]wire.Attribute) for { if l.Valid() { attributes[l.Key().Name()] = convertAttribute(l) } index++ if !list.Valid(index) { return &wire.Attributes{AttributeMap: attributes} } l = list.Label(index) } } func convertAttribute(l label.Label) wire.Attribute { switch key := l.Key().(type) { case *keys.Int: return wire.IntAttribute{IntValue: int64(key.From(l))} case *keys.Int8: return wire.IntAttribute{IntValue: int64(key.From(l))} case *keys.Int16: return wire.IntAttribute{IntValue: int64(key.From(l))} case *keys.Int32: return wire.IntAttribute{IntValue: int64(key.From(l))} case *keys.Int64: return wire.IntAttribute{IntValue: int64(key.From(l))} case *keys.UInt: return wire.IntAttribute{IntValue: int64(key.From(l))} case *keys.UInt8: return wire.IntAttribute{IntValue: int64(key.From(l))} case *keys.UInt16: return wire.IntAttribute{IntValue: int64(key.From(l))} case *keys.UInt32: return wire.IntAttribute{IntValue: int64(key.From(l))} case *keys.UInt64: return wire.IntAttribute{IntValue: int64(key.From(l))} case *keys.Float32: return wire.DoubleAttribute{DoubleValue: float64(key.From(l))} case *keys.Float64: return wire.DoubleAttribute{DoubleValue: key.From(l)} case *keys.Boolean: return wire.BoolAttribute{BoolValue: key.From(l)} case *keys.String: return wire.StringAttribute{StringValue: toTruncatableString(key.From(l))} case *keys.Error: return wire.StringAttribute{StringValue: toTruncatableString(key.From(l).Error())} case *keys.Value: return wire.StringAttribute{StringValue: toTruncatableString(fmt.Sprint(key.From(l)))} default: return wire.StringAttribute{StringValue: toTruncatableString(fmt.Sprintf("%T", key))} } } func convertEvents(events []core.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(ev core.Event) wire.TimeEvent { return wire.TimeEvent{ Time: convertTimestamp(ev.At()), Annotation: convertAnnotation(ev), } } func getAnnotationDescription(ev core.Event) (string, int) { l := ev.Label(0) if l.Key() != keys.Msg { return "", 0 } if msg := keys.Msg.From(l); msg != "" { return msg, 1 } l = ev.Label(1) if l.Key() != keys.Err { return "", 1 } if err := keys.Err.From(l); err != nil { return err.Error(), 2 } return "", 2 } func convertAnnotation(ev core.Event) *wire.Annotation { description, index := getAnnotationDescription(ev) if _, l := skipToValidLabel(ev, index); !l.Valid() && description == "" { return nil } return &wire.Annotation{ Description: toTruncatableString(description), Attributes: convertAttributes(ev, index), } }