push

Gopkg.toml

@@ -0,0 +1,38 @@
+# Gopkg.toml example
+#
+# Refer to https://golang.github.io/dep/docs/Gopkg.toml.html
+# for detailed Gopkg.toml documentation.
+#
+# required = ["github.com/user/thing/cmd/thing"]
+# ignored = ["github.com/user/project/pkgX", "bitbucket.org/user/project/pkgA/pkgY"]
+#
+# [[constraint]]
+#   name = "github.com/user/project"
+#   version = "1.0.0"
+#
+# [[constraint]]
+#   name = "github.com/user/project2"
+#   branch = "dev"
+#   source = "github.com/myfork/project2"
+#
+# [[override]]
+#   name = "github.com/x/y"
+#   version = "2.4.0"
+#
+# [prune]
+#   non-go = false
+#   go-tests = true
+#   unused-packages = true
+
+
+[[constraint]]
+  name = "github.com/opentracing/opentracing-go"
+  version = "1.1.0"
+
+[[constraint]]
+  name = "github.com/uber/jaeger-client-go"
+  version = "2.16.0"
+
+[prune]
+  go-tests = true
+  unused-packages = true

cmd/boshtaskdebugjaeger/main.go

@@ -0,0 +1,52 @@
+package main
+
+import (
+	"bufio"
+	"os"
+
+	"github.com/dpb587/boshdebugtracer/log"
+	"github.com/dpb587/boshdebugtracer/log/taskdebug"
+	"github.com/dpb587/boshdebugtracer/log/taskdebug/jaeger"
+	"github.com/dpb587/boshdebugtracer/observer/context"
+)
+
+func main() {
+	var err error
+
+	ctx := &context.Context{}
+	parsers := taskdebug.Parser
+
+	observer := jaeger.NewObserver(ctx)
+	observer.Begin()
+	defer observer.Commit()
+
+	var offset int64
+
+	scanner := bufio.NewScanner(os.Stdin)
+	for scanner.Scan() {
+		offset += 1
+
+		var l log.Line = log.RawLine{
+			RawLineOffset: offset,
+			RawLineData:   scanner.Text(),
+		}
+
+		for _, p := range parsers {
+			// fmt.Printf("IN: %#+v\n", l)
+			// fmt.Printf("TO: %#+v\n", p)
+			l, err = p.Parse(l)
+			if err != nil {
+				panic(err)
+			}
+		}
+
+		err := observer.Handle(l)
+		if err != nil {
+			panic(err)
+		}
+	}
+
+	if err := scanner.Err(); err != nil {
+		panic(err)
+	}
+}

log/interfaces.go

@@ -0,0 +1,10 @@
+package log
+
+type LineParser interface {
+	Parse(Line) (Line, error)
+}
+
+type Line interface {
+	LineOffset() int64
+	LineData() string
+}

log/raw.go

@@ -0,0 +1,16 @@
+package log
+
+type RawLine struct {
+	RawLineOffset int64
+	RawLineData   string
+}
+
+var _ Line = &RawLine{}
+
+func (m RawLine) LineOffset() int64 {
+	return m.RawLineOffset
+}
+
+func (m RawLine) LineData() string {
+	return m.RawLineData
+}

log/taskdebug/cpi_aws_rpc.go

@@ -0,0 +1,55 @@
+package taskdebug
+
+import (
+	"regexp"
+	"strconv"
+	"strings"
+	"time"
+
+	"github.com/dpb587/boshdebugtracer/log"
+)
+
+var CPIAWSRPCParser = cpiAWSRPCParser{}
+
+type cpiAWSRPCParser struct{}
+
+type CPIAWSRPCMessage struct {
+	RawMessage
+
+	Correlation   string
+	Duration      time.Duration
+	Payload       string
+	PayloadMethod string
+}
+
+// [Aws::EC2::Client 200 1.069542 0 retries] run_instances(...
+var cpiAWSRPCOneRE = regexp.MustCompile(`^\[Aws::EC2::Client (\d+) ([\d\.]+) (\d) retries\] (.+)$`)
+
+func (p cpiAWSRPCParser) Parse(inU log.Line) (log.Line, error) {
+	in, ok := inU.(RawMessage)
+	if !ok {
+		return inU, nil
+	}
+
+	if in.Component != "ExternalCpiLog" {
+		return inU, nil
+	}
+
+	if m := cpiAWSRPCOneRE.FindStringSubmatch(in.Message); len(m) > 0 {
+		out := CPIAWSRPCMessage{
+			RawMessage:  in,
+			Correlation: in.Tags["req_id"],
+			Payload:     m[4],
+		}
+
+		out.PayloadMethod = strings.SplitN(out.Payload, "(", 2)[0]
+
+		if res, err := strconv.ParseFloat(m[2], 64); err == nil {
+			out.Duration = time.Duration(int64(res * 1000000))
+		}
+
+		return out, nil
+	}
+
+	return inU, nil
+}

log/taskdebug/external_cpi.go

@@ -0,0 +1,46 @@
+package taskdebug
+
+import (
+	"regexp"
+
+	"github.com/dpb587/boshdebugtracer/log"
+)
+
+var ExternalCPIParser = externalCPIParser{}
+
+type externalCPIParser struct{}
+
+type ExternalCPIMessage struct {
+	RawMessage
+
+	Correlation string
+	Event       string
+	Payload     string
+}
+
+// [external-cpi] [cpi-308955] request: {"method":"create_vm","arguments":[...
+var externalCPIOneRE = regexp.MustCompile(`^\[external-cpi\] \[(cpi-\d+)\] (request|response): (.+)(\s+with command:|, err:)\s.+$`)
+
+func (p externalCPIParser) Parse(inU log.Line) (log.Line, error) {
+	in, ok := inU.(RawMessage)
+	if !ok {
+		return inU, nil
+	}
+
+	if in.Component != "DirectorJobRunner" {
+		return inU, nil
+	}
+
+	if m := externalCPIOneRE.FindStringSubmatch(in.Message); len(m) > 0 {
+		out := ExternalCPIMessage{
+			RawMessage:  in,
+			Correlation: m[1],
+			Event:       m[2],
+			Payload:     m[3],
+		}
+
+		return out, nil
+	}
+
+	return inU, nil
+}

log/taskdebug/external_cpi_request.go

@@ -0,0 +1,57 @@
+package taskdebug
+
+import (
+	"encoding/json"
+
+	"github.com/dpb587/boshdebugtracer/log"
+)
+
+var ExternalCPIRequestParser = externalCPIRequestParser{}
+
+type externalCPIRequestParser struct{}
+
+type ExternalCPIRequestMessage struct {
+	ExternalCPIMessage
+
+	PayloadMethod string
+}
+
+func (p externalCPIRequestParser) Parse(inU log.Line) (log.Line, error) {
+	inU, err := ExternalCPIParser.Parse(inU)
+	if inU == nil || err != nil {
+		return inU, err
+	}
+
+	in, ok := inU.(ExternalCPIMessage)
+	if !ok {
+		return inU, nil
+	}
+
+	upstreamU, err := ExternalCPIParser.Parse(in)
+	if upstreamU == nil || err != nil {
+		return upstreamU, err
+	}
+
+	upstream := upstreamU.(ExternalCPIMessage)
+
+	if upstream.Event != "request" {
+		return upstream, nil
+	}
+
+	out := ExternalCPIRequestMessage{
+		ExternalCPIMessage: upstream,
+	}
+
+	var payload struct {
+		Method string `json:"method"`
+	}
+
+	err = json.Unmarshal([]byte(out.Payload), &payload)
+	if err != nil {
+		panic(err)
+	}
+
+	out.PayloadMethod = payload.Method
+
+	return out, nil
+}