fix: fix logging an signal handling (#339)

kill/kill.go

@@ -1,6 +1,7 @@
 package main
 
 import (
+	"log"
 	"os"
 	"strconv"
 	"syscall"
@@ -21,7 +22,7 @@ func mainE(pid int) error {
 	if err != nil {
 		return err
 	}
-	println("signaling pid with SIGTERM", "pid", pid)
+	log.Printf("signaling pid %d with SIGTERM\n", pid)
 	if err := p.Signal(syscall.SIGTERM); err != nil {
 		return err
 	}

runner/main.go

@@ -9,6 +9,7 @@ import (
 	"net/http/pprof"
 	"os"
 	"os/signal"
+	"syscall"
 
 	"github.com/prometheus/client_golang/prometheus/promhttp"
 
@@ -47,7 +48,7 @@ func init() {
 }
 
 func main() {
-	ctx, stop := signal.NotifyContext(context.Background())
+	ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGTERM)
 	defer stop()
 
 	start := func(f builtin.Process) error {

sdks/golang/crash.go

@@ -0,0 +1,9 @@
+package golang
+
+import "log"
+
+func HandleCrash() {
+	if r := recover(); r != nil {
+		log.Printf("recovered from crash: %v\n", r)
+	}
+}

sdks/golang/start.go

@@ -7,12 +7,13 @@ import (
 	"net"
 	"net/http"
 	"os/signal"
+	"syscall"
 )
 
 //go:generate sh gen.sh
 
 func Start(handler func(ctx context.Context, msg []byte) ([]byte, error)) {
-	ctx, stop := signal.NotifyContext(context.Background())
+	ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGTERM)
 	defer stop()
 	if err := StartWithContext(ctx, handler); err != nil {
 		panic(err)
@@ -47,12 +48,7 @@ func StartWithContext(ctx context.Context, handler func(ctx context.Context, msg
 	// https://medium.com/honestbee-tw-engineer/gracefully-shutdown-in-go-http-server-5f5e6b83da5a
 	httpServer := &http.Server{Addr: ":8080"}
 	go func() {
-		defer func() {
-			r := recover()
-			if r != nil {
-				println(r)
-			}
-		}()
+		defer HandleCrash()
 		if err := httpServer.ListenAndServe(); err != nil && err != http.ErrServerClosed {
 			panic(err)
 		}
@@ -64,12 +60,7 @@ func StartWithContext(ctx context.Context, handler func(ctx context.Context, msg
 	}
 	defer func() { _ = listener.Close() }()
 	go func() {
-		defer func() {
-			r := recover()
-			if r != nil {
-				println(r)
-			}
-		}()
+		defer HandleCrash()
 		if err := udsServer.Serve(listener); err != nil && err != http.ErrServerClosed {
 			panic(err)
 		}

test/db-e2e/db_source_test.go

@@ -98,6 +98,7 @@ func TestDBSource(t *testing.T) {
 		},
 	})
 
+	WaitForPipeline()
 	WaitForPod()
 
 	defer StartPortForward("db-source-insert-0")()

test/metric.go

@@ -56,7 +56,6 @@ func ExpectMetric(name string, matcher matcher, opts ...interface{}) {
 	for {
 		select {
 		case <-ctx.Done():
-			println(getMetrics(ctx, port))
 			panic(fmt.Errorf("failed to wait for metric named %q to be %s: %w", name, matcher, ctx.Err()))
 		default:
 			found := false