profiler: collect profiles concurrently

profiler/profile.go

@@ -272,7 +272,10 @@ func (p *profiler) deltaProfile(name string, delta *pprofutils.Delta, curData []
 		return nil, fmt.Errorf("delta prof parse: %v", err)
 	}
 	var deltaData []byte
-	if prevProf := p.prev[name]; prevProf == nil {
+	p.mu.Lock()
+	prevProf := p.prev[name]
+	p.mu.Unlock()
+	if prevProf == nil {
 		// First time deltaProfile gets called for a type, there is no prevProf. In
 		// this case we emit the current profile as a delta profile.
 		deltaData = curData
@@ -298,7 +301,9 @@ func (p *profiler) deltaProfile(name string, delta *pprofutils.Delta, curData []
 	}
 	// Keep the most recent profiles in memory for future diffing. This needs to
 	// be taken into account when enforcing memory limits going forward.
+	p.mu.Lock()
 	p.prev[name] = curProf
+	p.mu.Unlock()
 	return &profile{data: deltaData}, nil
 }
 

profiler/profiler.go

@@ -63,6 +63,7 @@ func Stop() {
 // profiler collects and sends preset profiles to the Datadog API at a given frequency
 // using a given configuration.
 type profiler struct {
+	mu         sync.Mutex
 	cfg        *config                      // profile configuration
 	out        chan batch                   // upload queue
 	uploadFunc func(batch) error            // defaults to (*profiler).upload; replaced in tests
@@ -211,6 +212,12 @@ func (p *profiler) run() {
 // an item.
 func (p *profiler) collect(ticker <-chan time.Time) {
 	defer close(p.out)
+	var (
+		// mu guards completed
+		mu        sync.Mutex
+		completed []*profile
+		wg        sync.WaitGroup
+	)
 	for {
 		select {
 		case <-ticker:
@@ -226,15 +233,22 @@ func (p *profiler) collect(ticker <-chan time.Time) {
 			}
 
 			for _, t := range p.enabledProfileTypes() {
-				profs, err := p.runProfile(t)
-				if err != nil {
-					log.Error("Error getting %s profile: %v; skipping.", t, err)
-					p.cfg.statsd.Count("datadog.profiler.go.collect_error", 1, append(p.cfg.tags, t.Tag()), 1)
-					continue
-				}
-				for _, prof := range profs {
-					bat.addProfile(prof)
-				}
+				wg.Add(1)
+				go func(t ProfileType) {
+					defer wg.Done()
+					profs, err := p.runProfile(t)
+					if err != nil {
+						log.Error("Error getting %s profile: %v; skipping.", t, err)
+						p.cfg.statsd.Count("datadog.profiler.go.collect_error", 1, append(p.cfg.tags, t.Tag()), 1)
+					}
+					mu.Lock()
+					defer mu.Unlock()
+					completed = append(completed, profs...)
+				}(t)
+			}
+			wg.Wait()
+			for _, prof := range completed {
+				bat.addProfile(prof)
 			}
 			p.enqueueUpload(bat)
 		case <-p.exit: