Add json output format (#168)

* Support output format as json
* Add -output.format flag to caller
* Add -input.format flag to trex
* Use travis-pro for coverage target
* Build generate-schemas in Dockerfile
* Rename HopAnnotation2 type and schema
* Make file extension configurable
* Set extension based on output format
* Generalize ptr and waitprobe options for traceroutes
* Update flags for ptr and waitprobe

Author: stephen-soltesz

.travis.yml

@@ -18,4 +18,4 @@ script:
 after_success:
 # Coveralls
 # Upload coverage information for unit tests.
-- $GOPATH/bin/goveralls -coverprofile=_coverage.cov -service=travis-ci
+- $GOPATH/bin/goveralls -coverprofile=_coverage.cov -service=travis-pro

Dockerfile

@@ -9,6 +9,9 @@ RUN go get -v . && \
     go install -v -ldflags "-X github.com/m-lab/go/prometheusx.GitShortCommit=$(git log -1 --format=%h)" . && \
     chmod -R a+rx /go/bin/traceroute-caller
 
+RUN go install -v ./cmd/generate-schemas && \
+    chmod -R a+rx /go/bin/generate-schemas
+
 # Build and install the tools that are called by traceroute-caller.
 FROM ubuntu:22.04 as build_tracers
 RUN apt-get update && \
@@ -38,6 +41,7 @@ RUN mkdir -p /var/empty && \
     chmod 555 /var/empty
 # Copy the statically-linked traceroute-caller binary.
 COPY --from=build_caller /go/bin/traceroute-caller /
+COPY --from=build_caller /go/bin/generate-schemas /
 # Copy the dynamically-linked scamper binary and its associated libraries.
 COPY --from=build_tracers /scamper /usr/local
 

caller.go

@@ -33,8 +33,12 @@ var (
 		Options: []string{"mda", "regular"},
 		Value:   "mda",
 	}
-	scamperTracelbPTR   = flag.Bool("scamper.tracelb-ptr", true, "mda traceroute option: Look up DNS pointer records for IP addresses.")
-	scamperTracelbW     = flag.Int("scamper.tracelb-W", 25, "mda traceroute option: Wait time in 1/100ths of seconds between probes (min 15, max 200).")
+	outputFormat = flagx.Enum{
+		Options: []string{"jsonl", "json"},
+		Value:   "jsonl",
+	}
+	scamperTracePTR     = flag.Bool("scamper.ptr", true, "traceroute option: Look up DNS pointer records for IP addresses.")
+	scamperTraceWait    = flag.Int("scamper.waitprobe", 25, "traceroute option: Wait time in 1/100ths of seconds between probes (min 15, max 200).")
 	tracerouteOutput    = flag.String("traceroute-output", "/var/spool/scamper1", "The path to store traceroute output.")
 	hopAnnotationOutput = flag.String("hopannotation-output", "/var/spool/hopannotation1", "The path to store hop annotation output.")
 	// Keeping IP cache flags capitalized for backward compatibility.
@@ -52,6 +56,7 @@ var (
 
 func init() {
 	flag.Var(&scamperTraceType, "scamper.trace-type", "Specify the type of traceroute (mda or regular) to run.")
+	flag.Var(&outputFormat, "output.format", "Specify the output format of traces (jsonl or json).")
 }
 
 func main() {
@@ -80,14 +85,13 @@ func main() {
 
 	// 1. The traceroute tool (scamper).
 	scamperCfg := tracer.ScamperConfig{
-		Binary:     *scamperBin,
-		OutputPath: *tracerouteOutput,
-		Timeout:    *scamperTimeout,
-		TraceType:  scamperTraceType.Value,
-	}
-	if scamperCfg.TraceType == "mda" {
-		scamperCfg.TracelbPTR = *scamperTracelbPTR
-		scamperCfg.TracelbWaitProbe = *scamperTracelbW
+		Binary:         *scamperBin,
+		OutputPath:     *tracerouteOutput,
+		Timeout:        *scamperTimeout,
+		TracePTR:       *scamperTracePTR,
+		TraceWaitProbe: *scamperTraceWait,
+		TraceType:      scamperTraceType.Value,
+		Extension:      outputFormat.Value,
 	}
 	scamper, err := tracer.NewScamper(scamperCfg)
 	if err != nil {
@@ -103,7 +107,7 @@ func main() {
 		ScanPeriod:   *ipcScanPeriod,
 	}
 	// 3. The traceroute parser.
-	newParser, err := parser.New(scamperTraceType.Value)
+	newParser, err := parser.New(scamperTraceType.Value, outputFormat.Value)
 	if err != nil {
 		logFatal(err)
 	}

caller_test.go

@@ -73,7 +73,7 @@ func TestMainFunc(t *testing.T) {
 	for _, arg := range []strFlag{
 		{"-scamper.bin", "/bin/echo"},
 		{"-scamper.trace-type", "mda"},
-		{"-scamper.tracelb-W", "15"},
+		{"-scamper.waitprobe", "15"},
 		{"-prometheusx.listen-address", ":0"},
 		{"-tcpinfo.eventsocket", sockPath},
 		{"-traceroute-output", testDir},
@@ -107,7 +107,7 @@ func TestMainEventSocket(t *testing.T) {
 	for _, arg := range []strFlag{
 		{"-scamper.bin", "/bin/echo"},
 		{"-scamper.trace-type", "mda"},
-		{"-scamper.tracelb-W", "15"},
+		{"-scamper.waitprobe", "15"},
 		{"-prometheusx.listen-address", ":0"},
 		{"-tcpinfo.eventsocket", ""}, // should cause failure
 		{"-traceroute-output", testDir},
@@ -133,7 +133,7 @@ func TestMainScamper(t *testing.T) {
 	ctx, cancel = context.WithCancel(context.Background())
 	for _, arg := range []strFlag{
 		{"-scamper.bin", "/bin/echo"},
-		{"-scamper.tracelb-W", "10"}, // should cause failure (15 <= valid <= 200)
+		{"-scamper.waitprobe", "10"}, // should cause failure (15 <= valid <= 200)
 		{"-prometheusx.listen-address", ":0"},
 		{"-tcpinfo.eventsocket", sockPath},
 		{"-traceroute-output", testDir},
@@ -160,7 +160,7 @@ func TestMainNewHandler(t *testing.T) {
 	for _, arg := range []strFlag{
 		{"-scamper.bin", "/bin/echo"},
 		{"-scamper.trace-type", "mda"},
-		{"-scamper.tracelb-W", "15"},
+		{"-scamper.waitprobe", "15"},
 		{"-prometheusx.listen-address", ":0"},
 		{"-tcpinfo.eventsocket", sockPath},
 		{"-traceroute-output", testDir},

cmd/generate-schemas/main.go

@@ -0,0 +1,46 @@
+package main
+
+import (
+	"flag"
+	"os"
+
+	"github.com/m-lab/go/cloud/bqx"
+	"github.com/m-lab/go/rtx"
+	"github.com/m-lab/traceroute-caller/hopannotation"
+	"github.com/m-lab/traceroute-caller/parser"
+
+	"cloud.google.com/go/bigquery"
+)
+
+var (
+	scamper2schema string
+	hop2schema     string
+)
+
+func init() {
+	flag.StringVar(&scamper2schema, "scamper2", "/var/spool/datatypes/scamper2.json", "filename to write scamper2 schema")
+	flag.StringVar(&hop2schema, "hopannotation2", "/var/spool/datatypes/hopannotation2.json", "filename to write hopannotation2 schema")
+}
+
+func main() {
+	flag.Parse()
+	// TODO(soltesz): parser.Schema1 does not natively support BigQuery schema inference.
+
+	// Generate and save hopannotation2 schema for autoloading.
+	hop2 := hopannotation.HopAnnotation2{}
+	sch, err := bigquery.InferSchema(hop2)
+	rtx.Must(err, "failed to generate hopannotation2 schema")
+	sch = bqx.RemoveRequired(sch)
+	b, err := sch.ToJSONFields()
+	rtx.Must(err, "failed to marshal schema")
+	os.WriteFile(hop2schema, b, 0o644)
+
+	// Generate and save scamper2 schema for autoloading.
+	row2 := parser.Scamper2{}
+	sch, err = bigquery.InferSchema(row2)
+	rtx.Must(err, "failed to generate scamper2 schema")
+	sch = bqx.RemoveRequired(sch)
+	b, err = sch.ToJSONFields()
+	rtx.Must(err, "failed to marshal schema")
+	os.WriteFile(scamper2schema, b, 0o644)
+}

cmd/trex/main.go

@@ -9,14 +9,19 @@ import (
 	"path/filepath"
 	"sort"
 
+	"github.com/m-lab/go/flagx"
 	"github.com/m-lab/traceroute-caller/parser"
 )
 
 var (
-	prComplete = flag.Bool("c", false, "print flow IDs and file names of traceroutes that completed (\"--\" for incomplete traceroutes)")
-	duration   = flag.Uint("d", 0, "print times and file names of traceroutes that took more than the specified duration")
-	verbose    = flag.Bool("v", false, "enable verbose mode (mostly for debugging)")
-	flagSet    = make(map[string]bool)
+	prComplete  = flag.Bool("c", false, "print flow IDs and file names of traceroutes that completed (\"--\" for incomplete traceroutes)")
+	duration    = flag.Uint("d", 0, "print times and file names of traceroutes that took more than the specified duration")
+	verbose     = flag.Bool("v", false, "enable verbose mode (mostly for debugging)")
+	flagSet     = make(map[string]bool)
+	inputFormat = flagx.Enum{
+		Options: []string{"jsonl", "json"},
+		Value:   "jsonl",
+	}
 
 	// Statistics printed before exiting.
 	nFilesFound   uint32 // files found
@@ -31,6 +36,10 @@ var (
 	totDuration   uint64 // total duration of all traceroutes
 )
 
+func init() {
+	flag.Var(&inputFormat, "input.format", "Specify the input format of traces (jsonl or json).")
+}
+
 // Hop defines a hop.
 type Hop struct {
 	addr     string    // hop's IP address in string format
@@ -145,7 +154,7 @@ func parseFile(fileName string) *parser.Scamper1 {
 		nReadErrors++
 		return nil
 	}
-	mdaParser, err := parser.New("mda")
+	mdaParser, err := parser.New("mda", inputFormat.Value)
 	if err != nil {
 		fmt.Fprintf(os.Stderr, "%v\n", err)
 		return nil

docker-compose.yaml

@@ -72,8 +72,8 @@ services:
       - -scamper.trace-type=mda
       - -traceroute-output=/local/scamper1
       - -scamper.timeout=30m
-      - -scamper.tracelb-W=15
-      - -scamper.tracelb-ptr=true
+      - -scamper.waitprobe=15
+      - -scamper.ptr=true
 
   trc-scamper2:
     profiles:
@@ -101,3 +101,6 @@ services:
       - -scamper.trace-type=regular
       - -traceroute-output=/local/scamper2
       - -scamper.timeout=10m
+      - -scamper.waitprobe=15
+      - -scamper.ptr=true
+      - -output.format=json

go.mod

@@ -1,45 +1,64 @@
 module github.com/m-lab/traceroute-caller
 
-go 1.21
+go 1.20
 
 require (
+	cloud.google.com/go/bigquery v1.49.0
 	github.com/go-test/deep v1.0.7
-	github.com/m-lab/go v0.1.66
+	github.com/m-lab/go v0.1.75
 	github.com/m-lab/tcp-info v1.5.3
 	github.com/m-lab/uuid-annotator v0.4.7
 	github.com/prometheus/client_golang v1.12.2
 )
 
 require (
-	cloud.google.com/go v0.102.0 // indirect
-	cloud.google.com/go/compute v1.6.1 // indirect
-	cloud.google.com/go/iam v0.3.0 // indirect
-	cloud.google.com/go/storage v1.22.1 // indirect
+	cloud.google.com/go v0.110.0 // indirect
+	cloud.google.com/go/compute v1.18.0 // indirect
+	cloud.google.com/go/compute/metadata v0.2.3 // indirect
+	cloud.google.com/go/iam v0.12.0 // indirect
+	cloud.google.com/go/storage v1.29.0 // indirect
+	github.com/andybalholm/brotli v1.0.4 // indirect
+	github.com/apache/arrow/go/v11 v11.0.0 // indirect
+	github.com/apache/thrift v0.16.0 // indirect
 	github.com/araddon/dateparse v0.0.0-20210429162001-6b43995a97de // indirect
 	github.com/beorn7/perks v1.0.1 // indirect
-	github.com/cespare/xxhash/v2 v2.1.2 // indirect
+	github.com/cespare/xxhash/v2 v2.2.0 // indirect
+	github.com/goccy/go-json v0.9.11 // indirect
 	github.com/golang/groupcache v0.0.0-20210331224755-41bb18bfe9da // indirect
 	github.com/golang/protobuf v1.5.2 // indirect
-	github.com/google/go-cmp v0.5.8 // indirect
+	github.com/golang/snappy v0.0.4 // indirect
+	github.com/google/flatbuffers v2.0.8+incompatible // indirect
+	github.com/google/go-cmp v0.5.9 // indirect
 	github.com/google/uuid v1.3.0 // indirect
-	github.com/googleapis/gax-go/v2 v2.4.0 // indirect
-	github.com/googleapis/go-type-adapters v1.0.0 // indirect
+	github.com/googleapis/enterprise-certificate-proxy v0.2.3 // indirect
+	github.com/googleapis/gax-go/v2 v2.7.0 // indirect
 	github.com/googleapis/google-cloud-go-testing v0.0.0-20210719221736-1c9a4c676720 // indirect
+	github.com/klauspost/asmfmt v1.3.2 // indirect
+	github.com/klauspost/compress v1.15.9 // indirect
+	github.com/klauspost/cpuid/v2 v2.0.9 // indirect
 	github.com/matttproud/golang_protobuf_extensions v1.0.1 // indirect
+	github.com/minio/asm2plan9s v0.0.0-20200509001527-cdd76441f9d8 // indirect
+	github.com/minio/c2goasm v0.0.0-20190812172519-36a3d3bbc4f3 // indirect
 	github.com/oschwald/geoip2-golang v1.7.0 // indirect
 	github.com/oschwald/maxminddb-golang v1.9.0 // indirect
+	github.com/pierrec/lz4/v4 v4.1.15 // indirect
 	github.com/prometheus/client_model v0.2.0 // indirect
 	github.com/prometheus/common v0.34.0 // indirect
 	github.com/prometheus/procfs v0.7.3 // indirect
-	go.opencensus.io v0.23.0 // indirect
+	github.com/zeebo/xxh3 v1.0.2 // indirect
+	go.opencensus.io v0.24.0 // indirect
+	golang.org/x/mod v0.6.0-dev.0.20220419223038-86c51ed26bb4 // indirect
 	golang.org/x/net v0.7.0 // indirect
-	golang.org/x/oauth2 v0.0.0-20220524215830-622c5d57e401 // indirect
+	golang.org/x/oauth2 v0.5.0 // indirect
+	golang.org/x/sync v0.1.0 // indirect
 	golang.org/x/sys v0.5.0 // indirect
 	golang.org/x/text v0.7.0 // indirect
-	golang.org/x/xerrors v0.0.0-20220517211312-f3a8303e98df // indirect
-	google.golang.org/api v0.81.0 // indirect
+	golang.org/x/tools v0.1.12 // indirect
+	golang.org/x/xerrors v0.0.0-20220907171357-04be3eba64a2 // indirect
+	google.golang.org/api v0.111.0 // indirect
 	google.golang.org/appengine v1.6.7 // indirect
-	google.golang.org/genproto v0.0.0-20220525015930-6ca3db687a9d // indirect
-	google.golang.org/grpc v1.46.2 // indirect
-	google.golang.org/protobuf v1.28.0 // indirect
+	google.golang.org/genproto v0.0.0-20230306155012-7f2fa6fef1f4 // indirect
+	google.golang.org/grpc v1.53.0 // indirect
+	google.golang.org/protobuf v1.28.1 // indirect
+	gopkg.in/yaml.v2 v2.4.0 // indirect
 )