claudevdm commented on code in PR #39484:
URL: https://github.com/apache/beam/pull/39484#discussion_r3679221325
##########
sdks/python/container/profiler.go:
##########
@@ -334,3 +354,185 @@ func needsProcessing(binInfo os.FileInfo, path string)
bool {
// Don't regenerate when there were no updates to the profile.
return binInfo.ModTime().After(info.ModTime())
}
+
+func monitorCoredumpsLoop(ctx context.Context, logger *tools.Logger, pcfg
*ProfilerConfig) {
+ // We require the core file pattern to be configured as
"/tmp/core.%e.%p" via pipeline
+ // options experiments (which is automatically set by the Python SDK).
This ensures
+ // that core dumps are written in /tmp/ with a prefix matching "core.".
+ coreDir := "/tmp"
+ interval := 5 * time.Second
+ if pcfg.PostprocessIntervalSec > 0 {
+ interval = time.Duration(pcfg.PostprocessIntervalSec) *
time.Second
+ }
+ logger.Printf(ctx, "Monitoring directory %s for core dumps matching
prefix core. every %v", coreDir, interval)
+
+ ticker := time.NewTicker(interval)
+ defer ticker.Stop()
+
+ for {
+ select {
+ case <-ctx.Done():
+ return
+ case <-ticker.C:
+ processNewCoredumps(ctx, logger, pcfg, coreDir)
+ if isProfilerDisengaged(pcfg) {
+ return
+ }
+ }
+ }
+}
+
+func processNewCoredumps(ctx context.Context, logger *tools.Logger, pcfg
*ProfilerConfig, coreDir string) {
+ files, err := os.ReadDir(coreDir)
+ if err != nil {
+ return
+ }
+
+ prefix := "core."
Review Comment:
Is it a bit broad to match on any file starting with core.? Maybe we can add
a bit more validation/regex?
##########
sdks/python/container/profiler.go:
##########
@@ -334,3 +354,185 @@ func needsProcessing(binInfo os.FileInfo, path string)
bool {
// Don't regenerate when there were no updates to the profile.
return binInfo.ModTime().After(info.ModTime())
}
+
+func monitorCoredumpsLoop(ctx context.Context, logger *tools.Logger, pcfg
*ProfilerConfig) {
+ // We require the core file pattern to be configured as
"/tmp/core.%e.%p" via pipeline
+ // options experiments (which is automatically set by the Python SDK).
This ensures
+ // that core dumps are written in /tmp/ with a prefix matching "core.".
+ coreDir := "/tmp"
+ interval := 5 * time.Second
+ if pcfg.PostprocessIntervalSec > 0 {
+ interval = time.Duration(pcfg.PostprocessIntervalSec) *
time.Second
+ }
+ logger.Printf(ctx, "Monitoring directory %s for core dumps matching
prefix core. every %v", coreDir, interval)
+
+ ticker := time.NewTicker(interval)
+ defer ticker.Stop()
+
+ for {
+ select {
+ case <-ctx.Done():
+ return
+ case <-ticker.C:
+ processNewCoredumps(ctx, logger, pcfg, coreDir)
+ if isProfilerDisengaged(pcfg) {
+ return
Review Comment:
Is this a race with the 2s file-age guard in processNewCoredumps? With
--profiler_stop_after_crash.
Should we do a final synchronous sweep here that waits out the age guard
(sleep >2s, re-run processNewCoredumps, sync results to GCS) before returning?
##########
sdks/python/container/profiler.go:
##########
@@ -334,3 +354,185 @@ func needsProcessing(binInfo os.FileInfo, path string)
bool {
// Don't regenerate when there were no updates to the profile.
return binInfo.ModTime().After(info.ModTime())
}
+
+func monitorCoredumpsLoop(ctx context.Context, logger *tools.Logger, pcfg
*ProfilerConfig) {
+ // We require the core file pattern to be configured as
"/tmp/core.%e.%p" via pipeline
+ // options experiments (which is automatically set by the Python SDK).
This ensures
+ // that core dumps are written in /tmp/ with a prefix matching "core.".
+ coreDir := "/tmp"
+ interval := 5 * time.Second
+ if pcfg.PostprocessIntervalSec > 0 {
+ interval = time.Duration(pcfg.PostprocessIntervalSec) *
time.Second
+ }
+ logger.Printf(ctx, "Monitoring directory %s for core dumps matching
prefix core. every %v", coreDir, interval)
+
+ ticker := time.NewTicker(interval)
+ defer ticker.Stop()
+
+ for {
+ select {
+ case <-ctx.Done():
+ return
+ case <-ticker.C:
+ processNewCoredumps(ctx, logger, pcfg, coreDir)
Review Comment:
Should we syncProfilesToGCS here if the write was success?
##########
sdks/python/container/profiler.go:
##########
@@ -334,3 +354,185 @@ func needsProcessing(binInfo os.FileInfo, path string)
bool {
// Don't regenerate when there were no updates to the profile.
return binInfo.ModTime().After(info.ModTime())
}
+
+func monitorCoredumpsLoop(ctx context.Context, logger *tools.Logger, pcfg
*ProfilerConfig) {
+ // We require the core file pattern to be configured as
"/tmp/core.%e.%p" via pipeline
+ // options experiments (which is automatically set by the Python SDK).
This ensures
+ // that core dumps are written in /tmp/ with a prefix matching "core.".
+ coreDir := "/tmp"
+ interval := 5 * time.Second
Review Comment:
Is the default meant to be 5 seconds? It seems the pipeline option defaults
to 60 sec
https://github.com/apache/beam/blob/58bac320ebd2601e6b66261b5e40a72d59161cff/sdks/python/apache_beam/options/pipeline_options.py#L1724-L1726
So the 5 seconds will only survive if someone explicitly sets the pipeline
option to 0, otherwise it will default to 600 sec?
Is 600 too long to catch segfault on startup?
##########
sdks/python/container/boot.go:
##########
Review Comment:
The last crash's core dump seems to depend on the ticker firing before
Fatalf exits the container (default interval is 600s, and fresh cores <2s old
are skipped).
Should we do one explicit core-dump sweep + GCS sync right before failing
the container, so the crash that killed the worker is always analyzed
regardless of timer timing?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
