Create separate worker usage data collection and move hardware emit there #1293
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,151 @@ | ||
| package internal | ||
|
|
||
| import ( | ||
| "context" | ||
| "fmt" | ||
| "github.com/shirou/gopsutil/cpu" | ||
| "github.com/uber-go/tally" | ||
| "go.uber.org/cadence/internal/common/metrics" | ||
| "go.uber.org/zap" | ||
| "runtime" | ||
| "sync" | ||
| "time" | ||
| ) | ||
|
|
||
| type ( | ||
| workerUsageCollector struct { | ||
| workerType string | ||
| cooldownTime time.Duration | ||
| logger *zap.Logger | ||
| ctx context.Context | ||
| shutdownCh chan struct{} | ||
| wg *sync.WaitGroup // graceful stop | ||
| cancel context.CancelFunc | ||
| metricsScope tally.Scope | ||
| emitOncePerHost oncePerHost | ||
| } | ||
|
|
||
| workerUsageCollectorOptions struct { | ||
| Enabled bool | ||
| Cooldown time.Duration | ||
| MetricsScope tally.Scope | ||
| WorkerType string | ||
| EmitOnce oncePerHost | ||
| } | ||
|
|
||
| hardwareUsage struct { | ||
| NumCPUCores int | ||
| CPUPercent float64 | ||
| NumGoRoutines int | ||
| TotalMemory float64 | ||
| MemoryUsedHeap float64 | ||
| MemoryUsedStack float64 | ||
| } | ||
|
|
||
| oncePerHost interface { | ||
| Do(func()) | ||
| } | ||
| ) | ||
|
|
||
| func newWorkerUsageCollector( | ||
| options workerUsageCollectorOptions, | ||
| logger *zap.Logger, | ||
| ) *workerUsageCollector { | ||
| if !options.Enabled { | ||
| return nil | ||
| } | ||
| ctx, cancel := context.WithCancel(context.Background()) | ||
| return &workerUsageCollector{ | ||
| workerType: options.WorkerType, | ||
| cooldownTime: options.Cooldown, | ||
| metricsScope: options.MetricsScope, | ||
| logger: logger, | ||
| ctx: ctx, | ||
| cancel: cancel, | ||
| wg: &sync.WaitGroup{}, | ||
| emitOncePerHost: options.EmitOnce, | ||
| shutdownCh: make(chan struct{}), | ||
| } | ||
| } | ||
|
|
||
| func (w *workerUsageCollector) Start() { | ||
| w.wg.Add(1) | ||
| go func() { | ||
| defer func() { | ||
| if p := recover(); p != nil { | ||
| w.metricsScope.Counter(metrics.WorkerUsageCollectorPanic).Inc(1) | ||
| topLine := fmt.Sprintf("WorkerUsageCollector panic for workertype: %v", w.workerType) | ||
| st := getStackTraceRaw(topLine, 7, 0) | ||
| w.logger.Error("WorkerUsageCollector panic.", | ||
| zap.String(tagPanicError, fmt.Sprintf("%v", p)), | ||
| zap.String(tagPanicStack, st)) | ||
| } | ||
| }() | ||
| defer w.wg.Done() | ||
|
There was a problem hiding this comment. there are a few things problematic about this goroutine closure
|
||
|
|
||
| w.wg.Add(1) | ||
| w.logger.Info(fmt.Sprintf("Going to start hardware collector for workertype: %v", w.workerType)) | ||
| go w.runHardwareCollector() | ||
|
|
||
| }() | ||
| return | ||
| } | ||
|
|
||
| func (w *workerUsageCollector) Stop() { | ||
| close(w.shutdownCh) | ||
| w.wg.Wait() | ||
| w.cancel() | ||
|
|
||
| } | ||
|
|
||
| func (w *workerUsageCollector) runHardwareCollector() { | ||
| defer w.wg.Done() | ||
| ticker := time.NewTicker(w.cooldownTime) | ||
| defer ticker.Stop() | ||
| w.emitOncePerHost.Do(func() { | ||
| w.logger.Info(fmt.Sprintf("Started worker usage collector for workertype: %v", w.workerType)) | ||
| for { | ||
| select { | ||
| case <-w.shutdownCh: | ||
| return | ||
| case <-ticker.C: | ||
| hardwareUsageData := w.collectHardwareUsage() | ||
| if w.metricsScope != nil { | ||
| w.emitHardwareUsage(hardwareUsageData) | ||
| } | ||
| } | ||
| } | ||
| }) | ||
| } | ||
|
|
||
| func (w *workerUsageCollector) collectHardwareUsage() hardwareUsage { | ||
| cpuPercent, err := cpu.Percent(0, false) | ||
| if err != nil { | ||
| w.logger.Warn("Failed to get cpu percent", zap.Error(err)) | ||
| } | ||
| cpuCores, err := cpu.Counts(false) | ||
| if err != nil { | ||
| w.logger.Warn("Failed to get number of cpu cores", zap.Error(err)) | ||
| } | ||
|
|
||
| var memStats runtime.MemStats | ||
| runtime.ReadMemStats(&memStats) | ||
| return hardwareUsage{ | ||
| NumCPUCores: cpuCores, | ||
| CPUPercent: cpuPercent[0], | ||
| NumGoRoutines: runtime.NumGoroutine(), | ||
| TotalMemory: float64(memStats.Sys), | ||
| MemoryUsedHeap: float64(memStats.HeapAlloc), | ||
| MemoryUsedStack: float64(memStats.StackInuse), | ||
| } | ||
| } | ||
|
|
||
| // emitHardwareUsage emits collected hardware usage metrics to metrics scope | ||
| func (w *workerUsageCollector) emitHardwareUsage(usage hardwareUsage) { | ||
| w.metricsScope.Gauge(metrics.NumCPUCores).Update(float64(usage.NumCPUCores)) | ||
| w.metricsScope.Gauge(metrics.CPUPercentage).Update(usage.CPUPercent) | ||
| w.metricsScope.Gauge(metrics.NumGoRoutines).Update(float64(usage.NumGoRoutines)) | ||
| w.metricsScope.Gauge(metrics.TotalMemory).Update(float64(usage.TotalMemory)) | ||
| w.metricsScope.Gauge(metrics.MemoryUsedHeap).Update(float64(usage.MemoryUsedHeap)) | ||
| w.metricsScope.Gauge(metrics.MemoryUsedStack).Update(float64(usage.MemoryUsedStack)) | ||
| } | ||
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Do we need to spawn a goroutine per worker? Why not ensure only 1 running?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Only the hardware emitting is once per host, all other metrics will be worker-specific. (e.g activity poll response vs. decision poll response)
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
For now I see only w.collectHardwareUsage() which will just spawn bunch of data into the same scope. I would suggest separating hardware emitter and worker specific metrics.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
That's current design, for each type of metrics based on their origin, I will create a separate gorountine for each of them. But they would be contained under a single workerusagecollector so that their result can be collected and sent in one place