Move trace result writing from tracer package to triggertracer package (#151)

* Separate trace collection from result writing
* Write trace results from triggertrace package
* Disable broken docker build step

Author: stephen-soltesz

Dockerfile

@@ -40,9 +40,12 @@ RUN mkdir -p /var/empty && \
 COPY --from=build_caller /go/bin/traceroute-caller /
 # Copy the dynamically-linked scamper binary and its associated libraries.
 COPY --from=build_tracers /scamper /usr/local
+
 # Install fast-mda-traceroute from PyPI.
 # We build pycaracal from source to avoid pulling precompiled binaries.
-RUN pip3 install --no-binary pycaracal --no-cache-dir --verbose fast-mda-traceroute==0.1.10
+# TODO(#152): Build failures as of 2022-12-16. Re-enable once needed.
+# RUN pip3 install --no-binary pycaracal --no-cache-dir --verbose fast-mda-traceroute==0.1.10
+
 # Run ldconfig to locate all new libraries and verify the tools we need
 # are available.
 RUN ldconfig && \

internal/ipcache/ipcache.go

@@ -13,7 +13,7 @@ import (
 // Tracer is the generic interface for all things that can perform a traceroute.
 type Tracer interface {
 	Trace(remoteIP, uuid string, t time.Time) ([]byte, error)
-	CachedTrace(uuid string, t time.Time, cachedTrace []byte) error
+	CachedTrace(uuid string, t time.Time, cachedTrace []byte) ([]byte, error)
 	DontTrace()
 }
 
@@ -98,8 +98,8 @@ func (ic *IPCache) FetchTrace(remoteIP, uuid string) ([]byte, error) {
 			ic.tracetool.DontTrace()
 			return nil, cachedTrace.err
 		}
-		_ = ic.tracetool.CachedTrace(uuid, time.Now(), cachedTrace.data)
-		return cachedTrace.data, nil
+		data, err := ic.tracetool.CachedTrace(uuid, time.Now(), cachedTrace.data)
+		return data, err
 	}
 	cachedTrace.data, cachedTrace.err = ic.tracetool.Trace(remoteIP, uuid, cachedTrace.timeStamp)
 	close(cachedTrace.dataReady)

internal/ipcache/ipcache_test.go

@@ -28,9 +28,9 @@ func (ft *fakeTracer) Trace(remoteIP, uuid string, t time.Time) ([]byte, error)
 	return []byte("fake traceroute data to " + remoteIP), nil
 }
 
-func (ft *fakeTracer) CachedTrace(uuid string, t time.Time, cachedTest []byte) error {
+func (ft *fakeTracer) CachedTrace(uuid string, t time.Time, cachedTest []byte) ([]byte, error) {
 	ft.nCachedTrace++
-	return nil
+	return cachedTest, nil
 }
 
 func (ft *fakeTracer) DontTrace() {
@@ -132,10 +132,10 @@ func (pt *pausingTracer) Trace(remoteIP, uuid string, t time.Time) ([]byte, erro
 	return []byte("fake traceroute data to " + remoteIP), nil
 }
 
-func (pt *pausingTracer) CachedTrace(uuid string, t time.Time, cachedTest []byte) error {
+func (pt *pausingTracer) CachedTrace(uuid string, t time.Time, cachedTest []byte) ([]byte, error) {
 	randomDelay()
 	atomic.AddInt64(&pt.successes, 1)
-	return nil
+	return cachedTest, nil
 }
 
 func (pt *pausingTracer) DontTrace() {

internal/triggertrace/triggertrace.go

@@ -50,6 +50,11 @@ type AnnotateAndArchiver interface {
 	WriteAnnotations(map[string]*annotator.ClientAnnotations, time.Time) []error
 }
 
+type TracerWriter interface {
+	ipcache.Tracer
+	WriteFile(uuid string, t time.Time, b []byte) error
+}
+
 // Handler implements the tcp-info/eventsocket.Handler's interface.
 type Handler struct {
 	Destinations     map[string]Destination // key is UUID
@@ -58,11 +63,12 @@ type Handler struct {
 	IPCache          FetchTracer
 	Parser           ParseTracer
 	HopAnnotator     AnnotateAndArchiver
+	Tracer           TracerWriter
 	done             chan struct{} // For testing.
 }
 
 // NewHandler returns a new instance of Handler.
-func NewHandler(ctx context.Context, tracetool ipcache.Tracer, ipcCfg ipcache.Config, newParser parser.TracerouteParser, haCfg hopannotation.Config) (*Handler, error) {
+func NewHandler(ctx context.Context, tracetool TracerWriter, ipcCfg ipcache.Config, newParser parser.TracerouteParser, haCfg hopannotation.Config) (*Handler, error) {
 	ipCache, err := ipcache.New(ctx, tracetool, ipcCfg)
 	if err != nil {
 		return nil, err
@@ -81,6 +87,7 @@ func NewHandler(ctx context.Context, tracetool ipcache.Tracer, ipcCfg ipcache.Co
 		IPCache:      ipCache,
 		Parser:       newParser,
 		HopAnnotator: hopCache,
+		Tracer:       tracetool,
 	}, nil
 }
 
@@ -145,13 +152,17 @@ func (h *Handler) traceAnnotateAndArchive(ctx context.Context, uuid string, dest
 		log.Printf("context %p: failed to parse traceroute output (error: %v)\n", ctx, err)
 		return
 	}
+	traceStartTime := parsedData.StartTime()
+	err = h.Tracer.WriteFile(uuid, traceStartTime, rawData)
+	if err != nil {
+		log.Printf("context %p: failed to write trace file for uuid: %s: (error: %v)\n", ctx, uuid, err)
+	}
+
 	hops := parsedData.ExtractHops()
 	if len(hops) == 0 {
 		log.Printf("context %p: failed to extract hops from traceroute %+v\n", ctx, string(rawData))
 		return
 	}
-
-	traceStartTime := parsedData.StartTime()
 	annotations, allErrs := h.HopAnnotator.Annotate(ctx, hops, traceStartTime)
 	if allErrs != nil {
 		log.Printf("context %p: failed to annotate some or all hops (errors: %+v)\n", ctx, allErrs)

internal/triggertrace/triggertrace_test.go

@@ -56,10 +56,14 @@ func (ft *fakeTracer) Trace(remoteIP, uuid string, t time.Time) ([]byte, error)
 	return content, nil
 }
 
-func (ft *fakeTracer) CachedTrace(uuid string, t time.Time, cachedTest []byte) error {
+func (ft *fakeTracer) WriteFile(uuid string, t time.Time, data []byte) error {
+	return nil
+}
+
+func (ft *fakeTracer) CachedTrace(uuid string, t time.Time, cachedTest []byte) ([]byte, error) {
 	defer func() { atomic.AddInt32(&ft.nCachedTraces, 1) }()
 	fmt.Printf("\nCachedTrace()\n")
-	return nil
+	return nil, nil
 }
 
 func (ft *fakeTracer) DontTrace() {

tracer/scamper.go

@@ -13,6 +13,11 @@ import (
 	"time"
 )
 
+var (
+	// ErrEmptyUUID is returned when a required UUID is empty.
+	ErrEmptyUUID = errors.New("uuid is empty")
+)
+
 // ScamperConfig contains configuration parameters of scamper.
 type ScamperConfig struct {
 	Binary           string
@@ -76,35 +81,43 @@ func NewScamper(cfg ScamperConfig) (*Scamper, error) {
 	}, nil
 }
 
+// WriteFile writes the given data to a file in the configured Scamper output
+// path using the given UUID and time.
+func (s *Scamper) WriteFile(uuid string, t time.Time, data []byte) error {
+	filename, err := generateFilename(s.outputPath, uuid, t)
+	if err != nil {
+		return err
+	}
+	return os.WriteFile(filename, data, 0644)
+}
+
 // Trace starts a new scamper process to run a traceroute based on the
 // traceroute type and saves it in a file.
 func (s *Scamper) Trace(remoteIP, uuid string, t time.Time) ([]byte, error) {
 	tracesInProgress.WithLabelValues("scamper").Inc()
 	defer tracesInProgress.WithLabelValues("scamper").Dec()
+	if uuid == "" {
+		return nil, ErrEmptyUUID
+	}
 	return s.trace(remoteIP, uuid, t)
 }
 
-// CachedTrace creates a traceroute from the traceroute cache and saves it in a file.
-func (s *Scamper) CachedTrace(uuid string, t time.Time, cachedTrace []byte) error {
-	filename, err := generateFilename(s.outputPath, uuid, t)
-	if err != nil {
-		log.Printf("failed to generate filename (error: %v)\n", err)
-		tracerCacheErrors.WithLabelValues("scamper", err.Error()).Inc()
-		return err
+// CachedTrace creates an updated traceroute using the given uuid and time based on the traceroute cache.
+func (s *Scamper) CachedTrace(uuid string, t time.Time, cachedTrace []byte) ([]byte, error) {
+	if uuid == "" {
+		return nil, ErrEmptyUUID
 	}
-
 	// Remove the first line of the cached traceroute.
 	split := bytes.Index(cachedTrace, []byte{'\n'})
 	if split <= 0 || split == len(cachedTrace) {
 		log.Printf("failed to split cached traceroute (split: %v)\n", split)
 		tracerCacheErrors.WithLabelValues("scamper", "badcache").Inc()
-		return errors.New("invalid cached traceroute")
+		return nil, errors.New("invalid cached traceroute")
 	}
 
 	// Create and add the first line to the cached traceroute.
 	newTrace := append(createMetaline(uuid, true, extractUUID(cachedTrace[:split])), cachedTrace[split+1:]...)
-	// Make the file readable so it won't be overwritten.
-	return os.WriteFile(filename, []byte(newTrace), 0444)
+	return []byte(newTrace), nil
 }
 
 // DontTrace is called when a previous traceroute that we were waiting for
@@ -117,23 +130,15 @@ func (*Scamper) DontTrace() {
 // command line to invoke scamper varies depending on the traceroute type
 // and its options.
 func (s *Scamper) trace(remoteIP, uuid string, t time.Time) ([]byte, error) {
-	// Make sure a directory path based on the current date exists,
-	// generate a filename to save in that directory, and create
-	// a buffer to hold traceroute data.
-	filename, err := generateFilename(s.outputPath, uuid, t)
-	if err != nil {
-		return nil, err
-	}
-
-	// Create a context, run a traceroute, and write the output to file.
+	// Create a context, run a traceroute, and return the output.
 	ctx, cancel := context.WithTimeout(context.Background(), s.timeout)
 	defer cancel()
 	cmd := []string{s.binary, "-o-", "-O", "json", "-I", fmt.Sprintf("%s %s", s.cmd, remoteIP)}
-	return traceAndWrite(ctx, "scamper", filename, cmd, uuid)
+	return traceWithMeta(ctx, "scamper", cmd, uuid)
 }
 
-// traceAndWrite runs a traceroute and writes the result.
-func traceAndWrite(ctx context.Context, label string, filename string, cmd []string, uuid string) ([]byte, error) {
+// traceWithMeta runs a traceroute and adds a metadata line to the result.
+func traceWithMeta(ctx context.Context, label string, cmd []string, uuid string) ([]byte, error) {
 	data, err := runCmd(ctx, label, cmd)
 	if err != nil {
 		return nil, err
@@ -147,8 +152,7 @@ func traceAndWrite(ctx context.Context, label string, filename string, cmd []str
 	// the buffer becomes too large, Write() will panic with ErrTooLarge.
 	_, _ = buff.Write(createMetaline(uuid, false, ""))
 	_, _ = buff.Write(data)
-	// Make the file readable so it won't be overwritten.
-	return buff.Bytes(), os.WriteFile(filename, buff.Bytes(), 0444)
+	return buff.Bytes(), nil
 }
 
 // runCmd runs the given command and returns its output.
@@ -188,7 +192,7 @@ func runCmd(ctx context.Context, label string, cmd []string) ([]byte, error) {
 // generateFilename creates the string filename for storing the data.
 func generateFilename(path, uuid string, t time.Time) (string, error) {
 	if uuid == "" {
-		return "", errors.New("uuid is empty")
+		return "", ErrEmptyUUID
 	}
 	dir, err := createDatePath(path, t)
 	if err != nil {

tracer/scamper_test.go

@@ -82,7 +82,7 @@ func TestEmptyUUID(t *testing.T) {
 	if err == nil || !strings.Contains(err.Error(), wantErr) {
 		t.Errorf("Trace() = %v, want %q", err, wantErr)
 	}
-	err = s.CachedTrace("", time.Now(), []byte("does not matter"))
+	_, err = s.CachedTrace("", time.Now(), []byte("does not matter"))
 	if err == nil || !strings.Contains(err.Error(), wantErr) {
 		t.Errorf("Trace() = %v, want %q", err, wantErr)
 	}
@@ -139,6 +139,11 @@ func TestTrace(t *testing.T) {
 		if strings.TrimSpace(got) != strings.TrimSpace(test.want) {
 			t.Errorf("Trace() = %q, want %q", strings.TrimSpace(got), strings.TrimSpace(test.want))
 		}
+		err = s.WriteFile(uuid, now, out)
+		if err != nil {
+			t.Errorf("WriteFile() = %v, want nil", err)
+			continue
+		}
 		// Make sure that the output was correctly written to file.
 		out, err = os.ReadFile(filename)
 		if err != nil {
@@ -175,16 +180,20 @@ func TestTraceWritesMeta(t *testing.T) {
 	faketime := time.Date(2019, time.April, 1, 3, 45, 51, 0, time.UTC)
 	prometheusx.GitShortCommit = "Fake Version"
 	wantUUID := "0123456789"
-	_, err = s.Trace("1.2.3.4", wantUUID, faketime)
+	out, err := s.Trace("1.2.3.4", wantUUID, faketime)
 	if err != nil {
 		t.Errorf("Trace() = %v, want nil", err)
 	}
+	err = s.WriteFile(wantUUID, faketime, out)
+	if err != nil {
+		t.Errorf("WriteFile() = %v, want nil", err)
+	}
 
 	// Unmarshal the first line of the output file.
-	b, err := os.ReadFile(tempdir + "/2019/04/01/20190401T034551Z_" + wantUUID + ".jsonl")
+	out, err = os.ReadFile(tempdir + "/2019/04/01/20190401T034551Z_" + wantUUID + ".jsonl")
 	rtx.Must(err, "failed to read file")
 	m := Metadata{}
-	lines := strings.Split(string(b), "\n")
+	lines := strings.Split(string(out), "\n")
 	if len(lines) < 2 {
 		t.Errorf("len(lines) = %d, want 2", len(lines))
 	}
@@ -228,16 +237,16 @@ func TestCachedTrace(t *testing.T) {
 	{"type":"tracelb", "version":"0.1", "userid":0, "method":"icmp-echo", "src":"::ffff:180.87.97.101", "dst":"::ffff:1.47.236.62", "start":{"sec":1566691298, "usec":476221, "ftime":"2019-08-25 00:01:38"}, "probe_size":60, "firsthop":1, "attempts":3, "confidence":95, "tos":0, "gaplimit":3, "wait_timeout":5, "wait_probe":250, "probec":0, "probec_max":3000, "nodec":0, "linkc":0}
 	{"type":"cycle-stop", "list_name":"/tmp/scamperctrl:51811", "id":1, "hostname":"ndt-plh7v", "stop_time":1566691298}`)
 
-	_ = s.CachedTrace(uuid, faketime, []byte("Broken cached traceroute"))
-	_, errInvalidTest := os.ReadFile(tempdir + "/2019/04/01/20190401T034551Z_" + uuid + ".jsonl")
-	if errInvalidTest == nil {
-		t.Error("CachedTrace() = nil, want error")
+	_, err = s.CachedTrace(uuid, faketime, []byte("Broken cached traceroute"))
+	if err == nil {
+		t.Error("CacheTrace() returned nil error, want error")
 	}
 
-	_ = s.CachedTrace(uuid, faketime, cachedTrace)
+	b, err := s.CachedTrace(uuid, faketime, cachedTrace)
+	if err != nil {
+		t.Errorf("CacheTrace() = %v, want nil", err)
+	}
 	// Unmarshal the first line of the output file.
-	b, err := os.ReadFile(tempdir + "/2019/04/01/20190401T034551Z_" + uuid + ".jsonl")
-	rtx.Must(err, "failed to read file")
 	m := Metadata{}
 	lines := strings.Split(string(b), "\n")
 	if len(lines) < 2 {