This is an automated email from the ASF dual-hosted git repository.
jrmccluskey pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/master by this push:
new 76290b83c50 Add logging at error to Bootloader logger (#27813)
76290b83c50 is described below
commit 76290b83c506e018255b9a8b164b339c1bee6137
Author: Jack McCluskey <34928439+jrmcclus...@users.noreply.github.com>
AuthorDate: Fri Aug 4 11:58:40 2023 -0400
Add logging at error to Bootloader logger (#27813)
* Add logging at error to Bootloader logger
* Use Beam logging package instead of base go
* Revert "Use Beam logging package instead of base go"
This reverts commit 91f9b33832f5bb2bf415b5b942807e7aadfa5fe2.
---
sdks/go/container/tools/logging.go | 5 +++++
sdks/go/container/tools/logging_test.go | 16 ++++++++++++++++
sdks/python/container/boot.go | 14 ++++++++++----
sdks/python/container/piputil.go | 3 +--
4 files changed, 32 insertions(+), 6 deletions(-)
diff --git a/sdks/go/container/tools/logging.go
b/sdks/go/container/tools/logging.go
index 012efdb376b..ced13b744d4 100644
--- a/sdks/go/container/tools/logging.go
+++ b/sdks/go/container/tools/logging.go
@@ -111,6 +111,11 @@ func (l *Logger) Warnf(ctx context.Context, format string,
args ...any) {
l.Log(ctx, fnpb.LogEntry_Severity_WARN, fmt.Sprintf(format, args...))
}
+// Errorf logs the message with Error severity.
+func (l *Logger) Errorf(ctx context.Context, format string, args ...any) {
+ l.Log(ctx, fnpb.LogEntry_Severity_ERROR, fmt.Sprintf(format, args...))
+}
+
// Fatalf logs the message with Critical severity, and then calls os.Exit(1).
func (l *Logger) Fatalf(ctx context.Context, format string, args ...any) {
l.Log(ctx, fnpb.LogEntry_Severity_CRITICAL, fmt.Sprintf(format,
args...))
diff --git a/sdks/go/container/tools/logging_test.go
b/sdks/go/container/tools/logging_test.go
index e33b3c075a6..8730a0fe9c1 100644
--- a/sdks/go/container/tools/logging_test.go
+++ b/sdks/go/container/tools/logging_test.go
@@ -58,6 +58,22 @@ func TestLogger(t *testing.T) {
t.Errorf("l.Printf(\"foo %%v\", \"bar\"): got severity
%v, want %v", got, want)
}
})
+ t.Run("SuccessfulLoggingAtError", func(t *testing.T) {
+ catcher := &logCatcher{}
+ l := &Logger{client: catcher}
+
+ l.Errorf(ctx, "failed to install dependency %v", "bar")
+
+ received := catcher.msgs[0].GetLogEntries()[0]
+
+ if got, want := received.Message, "failed to install dependency
bar"; got != want {
+ t.Errorf("l.Printf(\"foo %%v\", \"bar\"): got message
%q, want %q", got, want)
+ }
+
+ if got, want := received.Severity,
fnpb.LogEntry_Severity_ERROR; got != want {
+ t.Errorf("l.Errorf(\"failed to install dependency
%%v\", \"bar\"): got severity %v, want %v", got, want)
+ }
+ })
t.Run("backup path", func(t *testing.T) {
catcher := &logCatcher{}
l := &Logger{client: catcher}
diff --git a/sdks/python/container/boot.go b/sdks/python/container/boot.go
index 4f8c2d084ff..c67239f0564 100644
--- a/sdks/python/container/boot.go
+++ b/sdks/python/container/boot.go
@@ -180,7 +180,10 @@ func launchSDKProcess() error {
dir := filepath.Join(*semiPersistDir, "staged")
files, err := artifact.Materialize(ctx, *artifactEndpoint,
info.GetDependencies(), info.GetRetrievalToken(), dir)
if err != nil {
- return fmt.Errorf("failed to retrieve staged files: %v", err)
+ fmtErr := fmt.Errorf("failed to retrieve staged files: %v", err)
+ // Send error message to logging service before returning up
the call stack
+ logger.Errorf(ctx, fmtErr.Error())
+ return fmtErr
}
// TODO(herohde): the packages to install should be specified
explicitly. It
@@ -198,7 +201,10 @@ func launchSDKProcess() error {
}
if setupErr := installSetupPackages(fileNames, dir, requirementsFiles);
setupErr != nil {
- return fmt.Errorf("failed to install required packages: %v",
setupErr)
+ fmtErr := fmt.Errorf("failed to install required packages: %v",
setupErr)
+ // Send error message to logging service before returning up
the call stack
+ logger.Errorf(ctx, fmtErr.Error())
+ return fmtErr
}
// (3) Invoke python
@@ -242,7 +248,7 @@ func launchSDKProcess() error {
// have elapsed, i.e., as soon as all
subprocesses have returned from Wait().
time.Sleep(5 * time.Second)
if err := syscall.Kill(-pid, syscall.SIGKILL);
err == nil {
- logger.Printf(ctx, "Worker process %v
did not respond, killed it.", pid)
+ logger.Warnf(ctx, "Worker process %v
did not respond, killed it.", pid)
}
}(pid)
syscall.Kill(-pid, syscall.SIGTERM)
@@ -278,7 +284,7 @@ func launchSDKProcess() error {
// DoFns throwing exceptions.
errorCount += 1
if errorCount < 4 {
- logger.Printf(ctx, "Python
(worker %v) exited %v times: %v\nrestarting SDK process",
+ logger.Warnf(ctx, "Python
(worker %v) exited %v times: %v\nrestarting SDK process",
workerId, errorCount,
err)
} else {
logger.Fatalf(ctx, "Python
(worker %v) exited %v times: %v\nout of retries, failing container",
diff --git a/sdks/python/container/piputil.go b/sdks/python/container/piputil.go
index 28bd92a791b..720bf372c53 100644
--- a/sdks/python/container/piputil.go
+++ b/sdks/python/container/piputil.go
@@ -20,7 +20,6 @@ import (
"bytes"
"errors"
"fmt"
- "io/ioutil"
"log"
"os"
"os/exec"
@@ -120,7 +119,7 @@ func installExtraPackages(files []string,
extraPackagesFile, dir string) error {
}
// Found the manifest. Install extra packages.
- manifest, err := ioutil.ReadFile(filepath.Join(dir,
extraPackagesFile))
+ manifest, err := os.ReadFile(filepath.Join(dir,
extraPackagesFile))
if err != nil {
return fmt.Errorf("failed to read extra packages
manifest file: %v", err)
}
