merge both TestCrash and Interrupt into the receiver

Author: siscia

internal/gateway/logging.go

@@ -16,6 +16,7 @@ type LogLevel zerolog.Level
 const (
 	LogDebug LogLevel = LogLevel(zerolog.DebugLevel)
 	LogInfo  LogLevel = LogLevel(zerolog.InfoLevel)
+	LogWarn  LogLevel = LogLevel(zerolog.WarnLevel)
 	LogError LogLevel = LogLevel(zerolog.ErrorLevel)
 )
 

internal/gateway/receiver/mock_receiver.go

@@ -10,31 +10,27 @@ import (
 // MockReceiver is a mocked implementation of the Receiver interface, for testing
 // Can implement fault injection
 type MockReceiver struct {
+	CvmfsReceiver
 }
 
 // NewMockReceiver constructs a new MockReceiver object which implements the
 // Receiver interface
-func NewMockReceiver(ctx context.Context) (Receiver, error) {
-	return &MockReceiver{}, nil
+func NewMockReceiver(ctx context.Context, execPath string, args ...string) (Receiver, error) {
+	CvmfsReceiver, err := NewCvmfsReceiver(ctx, execPath, args...)
+	return &MockReceiver{*CvmfsReceiver}, err
 }
 
-// Quit command
-func (r *MockReceiver) Quit() error {
-	return nil
-}
-
-// Echo command
-func (r *MockReceiver) Echo() error {
+func (r *MockReceiver) Commit(leasePath, oldRootHash, newRootHash string, tag gw.RepositoryTag) error {
+	gw.LogC(r.CvmfsReceiver.ctx, "mock receiver", gw.LogWarn).
+		Str("leaserPath", leasePath).
+		Msg("Requested commit agains a testing mock, shortcircuit success")
 	return nil
 }
 
-// SubmitPayload command
 func (r *MockReceiver) SubmitPayload(leasePath string, payload io.Reader, digest string, headerSize int) error {
-	return nil
-}
-
-// Commit command
-func (r *MockReceiver) Commit(leasePath, oldRootHash, newRootHash string, tag gw.RepositoryTag) error {
+	gw.LogC(r.CvmfsReceiver.ctx, "mock receiver", gw.LogWarn).
+		Str("leaserPath", leasePath).
+		Msg("Requested submit agains a testing mock, shortcircuit success")
 	return nil
 }
 

internal/gateway/receiver/pool.go

@@ -56,6 +56,19 @@ func (p commitTask) Context() context.Context {
 	return p.ctx
 }
 
+type testCrashTask struct {
+	ctx       context.Context
+	replyChan chan<- error
+}
+
+func (p testCrashTask) Reply() chan<- error {
+	return p.replyChan
+}
+
+func (p testCrashTask) Context() context.Context {
+	return p.ctx
+}
+
 // Pool maintains a number of parallel receiver workers to service
 // payload submission and commit requests. Payload submissions are done in
 // parallel, using Config.NumReceivers workers, while only a single commit
@@ -111,6 +124,14 @@ func (p *Pool) CommitLease(ctx context.Context, leasePath, oldRootHash, newRootH
 	return result
 }
 
+// this is private, it is not enough to just make it private but it is a good start
+func (p *Pool) testCrash(ctx context.Context) error {
+	reply := make(chan error)
+	p.tasks <- testCrashTask{ctx, reply}
+	result := <-reply
+	return result
+}
+
 func worker(tasks <-chan task, pool *Pool, workerIdx int) {
 	gw.Log("worker_pool", gw.LogDebug).
 		Int("worker_id", workerIdx).
@@ -149,6 +170,9 @@ M:
 			case commitTask:
 				result = receiver.Commit(t.leasePath, t.oldRootHash, t.newRootHash, t.tag)
 				taskType = "commit"
+			case testCrashTask:
+				result = receiver.TestCrash()
+				taskType = "testcrash"
 			default:
 				task.Reply() <- fmt.Errorf("unknown task type")
 				return

internal/gateway/receiver/receiver.go

@@ -8,7 +8,6 @@ import (
 	"io"
 	"os"
 	"os/exec"
-	"strconv"
 	"strings"
 
 	gw "github.com/cvmfs/gateway/internal/gateway"
@@ -36,7 +35,8 @@ const (
 	receiverCheckToken    // Unused
 	receiverSubmitPayload
 	receiverCommit
-	receiverError // Unused
+	receiverError     // Unused
+	receiverTestCrash // Used only in testing
 )
 
 // Receiver contains the operations that "receiver" worker processes perform
@@ -47,15 +47,16 @@ type Receiver interface {
 	Commit(leasePath, oldRootHash, newRootHash string, tag gw.RepositoryTag) error
 	Interrupt() error // like Ctrl-C SIGTERM -2
 	// Kill() error // like Crtl-D SIGKILL -9
+	TestCrash() error
 }
 
 // NewReceiver is the factory method for Receiver types
-func NewReceiver(ctx context.Context, execPath string, mock bool) (Receiver, error) {
+func NewReceiver(ctx context.Context, execPath string, mock bool, args ...string) (Receiver, error) {
 	if mock {
-		return NewMockReceiver(ctx)
+		return NewMockReceiver(ctx, execPath, append(args, "-w ''")...)
 	}
 
-	return NewCvmfsReceiver(ctx, execPath)
+	return NewCvmfsReceiver(ctx, execPath, args...)
 }
 
 // CvmfsReceiver spawns an external cvmfs_receiver worker process
@@ -74,12 +75,14 @@ type ReceiverReply struct {
 }
 
 // NewCvmfsReceiver will spawn an external cvmfs_receiver worker process and wait for a command
-func NewCvmfsReceiver(ctx context.Context, execPath string) (*CvmfsReceiver, error) {
+func NewCvmfsReceiver(ctx context.Context, execPath string, args ...string) (*CvmfsReceiver, error) {
 	if _, err := os.Stat(execPath); os.IsNotExist(err) {
 		return nil, errors.Wrap(err, "worker process executable not found")
 	}
 
-	cmd := exec.Command(execPath, "-i", strconv.Itoa(3), "-o", strconv.Itoa(4))
+	cmdLine := []string{"-i", "3", "-o", "4"}
+	cmdLine = append(cmdLine, args...)
+	cmd := exec.Command(execPath, cmdLine...)
 
 	workerInRead, workerInWrite, err := os.Pipe()
 	if err != nil {
@@ -115,6 +118,12 @@ func NewCvmfsReceiver(ctx context.Context, execPath string) (*CvmfsReceiver, err
 		return nil, errors.Wrap(err, "could not start worker process")
 	}
 
+	// it is necessary to close this two files, otherwise, if the receiver crash,
+	// a read on the `workerOutRead` / `workerCmdOut` will hang forever.
+	// details: https://web.archive.org/web/20200429092830/https://redbeardlab.com/2020/04/29/on-linux-pipes-fork-and-passing-file-descriptors-to-other-process/
+	workerInRead.Close()
+	workerOutWrite.Close()
+
 	gw.LogC(ctx, "receiver", gw.LogDebug).
 		Str("command", "start").
 		Msg("worker process ready")
@@ -241,10 +250,21 @@ func (r *CvmfsReceiver) Interrupt() error {
 	return r.worker.Process.Signal(os.Interrupt)
 }
 
+// Method used only in testing, we provide an empty implementation here
+func (r *CvmfsReceiver) TestCrash() error {
+	reply, err := r.call(receiverTestCrash, nil, nil)
+	if err != nil {
+		return err
+	}
+	result := toReceiverError(reply)
+	return result
+}
+
 func (r *CvmfsReceiver) call(reqID receiverOp, msg []byte, payload io.Reader) ([]byte, error) {
 	if err := r.request(reqID, msg, payload); err != nil {
 		return nil, err
 	}
+
 	return r.reply()
 }
 
@@ -269,12 +289,18 @@ func (r *CvmfsReceiver) request(reqID receiverOp, msg []byte, payload io.Reader)
 func (r *CvmfsReceiver) reply() ([]byte, error) {
 	buf := make([]byte, 4)
 	if _, err := io.ReadFull(r.workerCmdOut, buf); err != nil {
+		if (err == io.EOF) || (err == io.ErrUnexpectedEOF) {
+			return nil, errors.Wrap(err, "possible that the receiver crashed")
+		}
 		return nil, errors.Wrap(err, "could not read reply size")
 	}
 	repSize := int32(binary.LittleEndian.Uint32(buf))
 
 	reply := make([]byte, repSize)
 	if _, err := io.ReadFull(r.workerCmdOut, reply); err != nil {
+		if (err == io.EOF) || (err == io.ErrUnexpectedEOF) {
+			return nil, errors.Wrap(err, "possible that the receiver crashed")
+		}
 		return nil, errors.Wrap(err, "could not read reply body")
 	}
 

internal/gateway/receiver/receiver_test.go

@@ -12,8 +12,59 @@ func TestReceiverCycle(t *testing.T) {
 	}
 	if err := receiver.Echo(); err != nil {
 		t.Fatalf("echo request failed: %v", err)
+
 	}
 	if err := receiver.Quit(); err != nil {
 		t.Fatalf("quit request failed: %v", err)
 	}
 }
+
+func TestReceiverOnCrashWeReturnError(t *testing.T) {
+	receiver, err := NewReceiver(context.TODO(), "/usr/bin/cvmfs_receiver", true)
+	if err != nil {
+		t.Fatalf("could not start receiver: %v", err)
+	}
+	if err := receiver.Echo(); err != nil {
+		t.Fatalf("echo request failed: %v", err)
+	}
+	err = receiver.TestCrash()
+	// note how we check the err being equal (==) and not different (!=) to nil
+	if err == nil {
+		t.Fatalf("crash request failed: %v", err)
+	}
+}
+
+func TestReceiverAfterCrashWeCanStillCallCommandAndTheyWillReturnAnError(t *testing.T) {
+	receiver, err := NewReceiver(context.TODO(), "/usr/bin/cvmfs_receiver", true)
+	if err != nil {
+		t.Fatalf("could not start receiver: %v", err)
+	}
+	if err := receiver.Echo(); err != nil {
+		t.Fatalf("echo request failed: %v", err)
+	}
+
+	receiver.TestCrash()
+
+	if receiver.Echo() == nil {
+		t.Fatalf("echo after crash didn't return nil")
+	}
+
+	if receiver.Quit() == nil {
+		t.Fatalf("quit after crash didn't return nil")
+	}
+}
+
+// reduntat test, but it mimic a problem we found in production.
+// after a crash the .Quit() was hanging
+func TestReceiverAfterCrashQuiteDoesNotHang(t *testing.T) {
+	receiver, err := NewReceiver(context.TODO(), "/usr/bin/cvmfs_receiver", true)
+	if err != nil {
+		t.Fatalf("could not start receiver: %v", err)
+	}
+
+	receiver.TestCrash()
+
+	if receiver.Quit() == nil {
+		t.Fatalf("quit after crash didn't return nil")
+	}
+}