Stac 21470: fix race

cmd/root.go

@@ -75,6 +75,7 @@ Additional targets are configured via PUT/DELETE on the '/targets?url=<endpoint>
 	cmd.Flags().Int("max-queued-requests", 500, "Maximum amount of requests queued per mirror.")                                                         //nolint:gomnd
 	cmd.Flags().Int("main-target-delay-ms", 0, "Delay delivery to main target, allowing slower mirrors to keep up and increase discovered parallelism.") //nolint:gomnd
 	cmd.Flags().Int("retry-after", 1, "After 5 successive failures a target is temporarily disabled, it will be retried after this many minutes.")
+	cmd.Flags().Bool("enable-pprof", false, "Enable pprof.")
 	cmd.Flags().StringSlice("mirror", []string{}, "Start with mirroring traffic to provided targets")
 
 	return cmd

internal/config/config.go

@@ -17,6 +17,7 @@ type Config struct {
 	Mirrors                  []string `yaml:"mirror"`
 	MaxQueuedRequests        int      `yaml:"max-queued-requests" default:"500"`
 	MainTargetDelayMs        int      `yaml:"main-target-delay-ms" default:"0"`
+	EnablePProf              bool     `yaml:"enable-pprof" default:"false"`
 }
 
 func (s *Config) UnmarshalYAML(unmarshal func(interface{}) error) error {

internal/mirror/reflector.go

@@ -1,10 +1,10 @@
 package mirror
 
 import (
+	"github.com/rb3ckers/trafficmirror/internal/config"
 	"log"
 	"sync"
-
-	"github.com/rb3ckers/trafficmirror/internal/config"
+	"time"
 )
 
 type Reflector struct {
@@ -78,7 +78,7 @@ func (r *Reflector) AddMirrors(urls []string, persistent bool) {
 
 	for _, url := range urls {
 		log.Printf("Adding '%s' to mirror list.", url)
-		r.mirrors[url] = NewMirror(url, r.config, r.MirrorFailureChan, persistent, r.templateSendQueue)
+		r.mirrors[url] = NewMirror(url, r.config, r.MirrorFailureChan, persistent, r.templateSendQueue.Clone())
 	}
 }
 
@@ -93,14 +93,27 @@ func (r *Reflector) RemoveMirrors(urls []string) {
 }
 
 func (r *Reflector) ListMirrors() []*MirrorStatus {
-	targets := make([]*MirrorStatus, len(r.mirrors))
+	r.Lock()
+	defer r.Unlock()
+
+	targets := make([]*MirrorStatus, len(r.mirrors)+1)
 	i := 0
 
 	for _, target := range r.mirrors {
 		targets[i] = target.GetStatus()
 		i++
 	}
 
+	epoch, requests := r.templateSendQueue.QueueStatus()
+
+	targets[i] = &MirrorStatus{
+		State:          StateAlive,
+		FailingSince:   time.Time{},
+		URL:            "internal-reflector",
+		QueuedRequests: requests,
+		Epoch:          epoch,
+	}
+
 	return targets
 }
 

internal/mirror/sendqueue.go

@@ -30,6 +30,9 @@ func MakeSendQueue(maxQueueSize int) *SendQueue {
 }
 
 func (s *SendQueue) Clone() *SendQueue {
+	s.Lock()
+	defer s.Unlock()
+
 	completedCopied := make(map[uint64]interface{}, 0)
 	for c, _ := range s.epochsCompleted {
 		completedCopied[c] = nil

internal/proxy/proxy.go

@@ -7,6 +7,7 @@ import (
 	"io/ioutil"
 	"log"
 	"net/http"
+	"net/http/pprof"
 	"net/url"
 	"strings"
 	"sync"
@@ -189,5 +190,13 @@ func (p *Proxy) setupTargetsMux(targetsMux *http.ServeMux) error {
 		targetsMux.HandleFunc("/"+p.cfg.TargetsEndpoint, p.mirrorsHandler)
 	}
 
+	if p.cfg.EnablePProf {
+		targetsMux.HandleFunc("/debug/pprof/", pprof.Index)
+		targetsMux.HandleFunc("/debug/pprof/cmdline", pprof.Cmdline)
+		targetsMux.HandleFunc("/debug/pprof/profile", pprof.Profile)
+		targetsMux.HandleFunc("/debug/pprof/symbol", pprof.Symbol)
+		targetsMux.HandleFunc("/debug/pprof/trace", pprof.Trace)
+	}
+
 	return nil
 }