[FAB-13238] handle container exit during launch

When a chaincode container has successfully started, it is possible for
the chaincode inside the container to fail before registration is
complete. Without this change, the container termination is not observed
by the peer and so, instead of failing quickly, the launch must time
out.

This commit watches for container termination during launch and will
report container exit before registration. This results in better
failure semantics and easier diagnostics.

Change-Id: If3071ee9e406b55a54d4d49e69dac9796e565e5d
Signed-off-by: Matthew Sykes <sykesmat@us.ibm.com>

Author: sykesm

core/chaincode/chaincode_support.go

@@ -27,6 +27,7 @@ import (
 type Runtime interface {
 	Start(ccci *ccprovider.ChaincodeContainerInfo, codePackage []byte) error
 	Stop(ccci *ccprovider.ChaincodeContainerInfo) error
+	Wait(ccci *ccprovider.ChaincodeContainerInfo) (int, error)
 }
 
 // Launcher is used to launch chaincode runtimes.

core/chaincode/container_runtime.go

@@ -99,6 +99,33 @@ func (c *ContainerRuntime) Stop(ccci *ccprovider.ChaincodeContainerInfo) error {
 	return nil
 }
 
+// Wait waits for the container runtime to terminate.
+func (c *ContainerRuntime) Wait(ccci *ccprovider.ChaincodeContainerInfo) (int, error) {
+	type result struct {
+		exitCode int
+		err      error
+	}
+
+	resultCh := make(chan result, 1)
+	wcr := container.WaitContainerReq{
+		CCID: ccintf.CCID{
+			Name:    ccci.Name,
+			Version: ccci.Version,
+		},
+		Exited: func(exitCode int, err error) {
+			resultCh <- result{exitCode: exitCode, err: err}
+			close(resultCh)
+		},
+	}
+
+	if err := c.Processor.Process(ccci.ContainerType, wcr); err != nil {
+		return -1, err
+	}
+	r := <-resultCh
+
+	return r.exitCode, r.err
+}
+
 const (
 	// Mutual TLS auth client key and cert paths in the chaincode container
 	TLSClientKeyPath      string = "/etc/hyperledger/fabric/client.key"

core/chaincode/container_runtime_test.go

@@ -280,3 +280,37 @@ func TestContainerRuntimeStopErrors(t *testing.T) {
 		assert.NoError(t, err)
 	}
 }
+
+func TestContainerRuntimeWait(t *testing.T) {
+	fakeProcessor := &mock.Processor{}
+	fakeProcessor.ProcessStub = func(containerType string, req container.VMCReq) error {
+		waitReq := req.(container.WaitContainerReq)
+		waitReq.Exited(0, nil)
+		return nil
+	}
+	cr := &chaincode.ContainerRuntime{
+		Processor: fakeProcessor,
+	}
+
+	ccci := &ccprovider.ChaincodeContainerInfo{
+		Type:          pb.ChaincodeSpec_GOLANG.String(),
+		Name:          "chaincode-id-name",
+		Version:       "chaincode-version",
+		ContainerType: "container-type",
+	}
+
+	exitCode, err := cr.Wait(ccci)
+	assert.NoError(t, err)
+	assert.Equal(t, 0, exitCode)
+
+	assert.Equal(t, 1, fakeProcessor.ProcessCallCount())
+	vmType, req := fakeProcessor.ProcessArgsForCall(0)
+	assert.Equal(t, vmType, "container-type")
+	waitReq, ok := req.(container.WaitContainerReq)
+	assert.True(t, ok)
+	assert.Equal(t, ccintf.CCID{Name: "chaincode-id-name", Version: "chaincode-version"}, waitReq.CCID)
+
+	fakeProcessor.ProcessReturns(errors.New("moles-and-trolls"))
+	_, err = cr.Wait(ccci)
+	assert.EqualError(t, err, "moles-and-trolls")
+}

core/chaincode/mock/runtime.go

@@ -41,8 +41,8 @@ func (r *RuntimeLauncher) Launch(ccci *ccprovider.ChaincodeContainerInfo) error
 
 	startTime := time.Now()
 	cname := ccci.Name + ":" + ccci.Version
-	launchState, started := r.Registry.Launching(cname)
-	if !started {
+	launchState, alreadyStarted := r.Registry.Launching(cname)
+	if !alreadyStarted {
 		startFailCh = make(chan error, 1)
 		timeoutCh = time.NewTimer(r.StartupTimeout).C
 
@@ -54,7 +54,13 @@ func (r *RuntimeLauncher) Launch(ccci *ccprovider.ChaincodeContainerInfo) error
 		go func() {
 			if err := r.Runtime.Start(ccci, codePackage); err != nil {
 				startFailCh <- errors.WithMessage(err, "error starting container")
+				return
 			}
+			exitCode, err := r.Runtime.Wait(ccci)
+			if err != nil {
+				launchState.Notify(errors.Wrap(err, "failed to wait on container exit"))
+			}
+			launchState.Notify(errors.Errorf("container exited with %d", exitCode))
 		}()
 	}
 
@@ -64,19 +70,15 @@ func (r *RuntimeLauncher) Launch(ccci *ccprovider.ChaincodeContainerInfo) error
 		err = errors.WithMessage(launchState.Err(), "chaincode registration failed")
 	case err = <-startFailCh:
 		launchState.Notify(err)
-		r.Metrics.LaunchFailures.With(
-			"chaincode", cname,
-		).Add(1)
+		r.Metrics.LaunchFailures.With("chaincode", cname).Add(1)
 	case <-timeoutCh:
 		err = errors.Errorf("timeout expired while starting chaincode %s for transaction", cname)
 		launchState.Notify(err)
-		r.Metrics.LaunchTimeouts.With(
-			"chaincode", cname,
-		).Add(1)
+		r.Metrics.LaunchTimeouts.With("chaincode", cname).Add(1)
 	}
 
 	success := true
-	if err != nil && !started {
+	if err != nil && !alreadyStarted {
 		success = false
 		chaincodeLogger.Debugf("stopping due to error while launching: %+v", err)
 		defer r.Registry.Deregister(cname)

core/chaincode/runtime_launcher.go

@@ -28,6 +28,7 @@ var _ = Describe("RuntimeLauncher", func() {
 		fakeLaunchDuration  *metricsfakes.Histogram
 		fakeLaunchFailures  *metricsfakes.Counter
 		fakeLaunchTimeouts  *metricsfakes.Counter
+		exitedCh            chan int
 
 		ccci *ccprovider.ChaincodeContainerInfo
 
@@ -44,6 +45,11 @@ var _ = Describe("RuntimeLauncher", func() {
 			launchState.Notify(nil)
 			return nil
 		}
+		exitedCh = make(chan int)
+		waitExitCh := exitedCh // shadow to avoid race
+		fakeRuntime.WaitStub = func(*ccprovider.ChaincodeContainerInfo) (int, error) {
+			return <-waitExitCh, nil
+		}
 
 		fakePackageProvider = &mock.PackageProvider{}
 		fakePackageProvider.GetChaincodeCodePackageReturns([]byte("code-package"), nil)
@@ -77,6 +83,10 @@ var _ = Describe("RuntimeLauncher", func() {
 		}
 	})
 
+	AfterEach(func() {
+		close(exitedCh)
+	})
+
 	It("registers the chaincode as launching", func() {
 		err := runtimeLauncher.Launch(ccci)
 		Expect(err).NotTo(HaveOccurred())
@@ -172,6 +182,34 @@ var _ = Describe("RuntimeLauncher", func() {
 		})
 	})
 
+	Context("when the contaienr terminates before registration", func() {
+		BeforeEach(func() {
+			fakeRuntime.StartReturns(nil)
+			fakeRuntime.WaitReturns(-99, nil)
+		})
+
+		It("returns an error", func() {
+			err := runtimeLauncher.Launch(ccci)
+			Expect(err).To(MatchError("chaincode registration failed: container exited with -99"))
+		})
+
+		It("attempts to stop the runtime", func() {
+			runtimeLauncher.Launch(ccci)
+
+			Expect(fakeRuntime.StopCallCount()).To(Equal(1))
+			ccciArg := fakeRuntime.StopArgsForCall(0)
+			Expect(ccciArg).To(Equal(ccci))
+		})
+
+		It("deregisters the chaincode", func() {
+			runtimeLauncher.Launch(ccci)
+
+			Expect(fakeRegistry.DeregisterCallCount()).To(Equal(1))
+			cname := fakeRegistry.DeregisterArgsForCall(0)
+			Expect(cname).To(Equal("chaincode-name:chaincode-version"))
+		})
+	})
+
 	Context("when handler registration fails", func() {
 		BeforeEach(func() {
 			fakeRuntime.StartStub = func(*ccprovider.ChaincodeContainerInfo, []byte) error {

core/chaincode/runtime_launcher_test.go

@@ -30,6 +30,7 @@ type Builder interface {
 type VM interface {
 	Start(ccid ccintf.CCID, args []string, env []string, filesToUpload map[string][]byte, builder Builder) error
 	Stop(ccid ccintf.CCID, timeout uint, dontkill bool, dontremove bool) error
+	Wait(ccid ccintf.CCID) (int, error)
 	HealthCheck(context.Context) error
 }
 
@@ -172,6 +173,31 @@ func (si StopContainerReq) GetCCID() ccintf.CCID {
 	return si.CCID
 }
 
+//go:generate counterfeiter -o mock/exitedfunc.go --fake-name ExitedFunc ExitedFunc
+
+// ExitedFunc is the prototype for the function called when a container exits.
+type ExitedFunc func(exitCode int, err error)
+
+// WaitContainerReq provides the chaincode ID of the container to wait on and a
+// callback to call upon chaincode termination.
+type WaitContainerReq struct {
+	CCID   ccintf.CCID
+	Exited ExitedFunc
+}
+
+func (w WaitContainerReq) Do(v VM) error {
+	exited := w.Exited
+	go func() {
+		exitCode, err := v.Wait(w.CCID)
+		exited(exitCode, err)
+	}()
+	return nil
+}
+
+func (w WaitContainerReq) GetCCID() ccintf.CCID {
+	return w.CCID
+}
+
 func (vmc *VMController) Process(vmtype string, req VMCReq) error {
 	v := vmc.newVM(vmtype)
 	ccid := req.GetCCID()

core/container/controller.go

@@ -13,7 +13,11 @@ import (
 	"github.com/hyperledger/fabric/core/chaincode/platforms"
 	"github.com/hyperledger/fabric/core/chaincode/platforms/golang"
 	"github.com/hyperledger/fabric/core/container"
+	"github.com/hyperledger/fabric/core/container/ccintf"
+	"github.com/hyperledger/fabric/core/container/mock"
 	pb "github.com/hyperledger/fabric/protos/peer"
+	. "github.com/onsi/gomega"
+	"github.com/pkg/errors"
 	"github.com/stretchr/testify/assert"
 )
 
@@ -32,3 +36,32 @@ func TestVM_GetChaincodePackageBytes(t *testing.T) {
 	assert.Error(t, err,
 		"GetChaincodePackageBytes did not return error when chaincode ID is nil")
 }
+
+func TestWaitContainerReq(t *testing.T) {
+	gt := NewGomegaWithT(t)
+
+	exited := &mock.ExitedFunc{}
+	done := make(chan struct{})
+	exited.Stub = func(int, error) { close(done) }
+
+	req := container.WaitContainerReq{
+		CCID:   ccintf.CCID{Name: "the-name", Version: "the-version"},
+		Exited: exited.Spy,
+	}
+	gt.Expect(req.GetCCID()).To(Equal(ccintf.CCID{Name: "the-name", Version: "the-version"}))
+
+	fakeVM := &mock.VM{}
+	fakeVM.WaitReturns(99, errors.New("boing-boing"))
+
+	err := req.Do(fakeVM)
+	gt.Expect(err).NotTo(HaveOccurred())
+	gt.Eventually(done).Should(BeClosed())
+
+	gt.Expect(fakeVM.WaitCallCount()).To(Equal(1))
+	ccid := fakeVM.WaitArgsForCall(0)
+	gt.Expect(ccid).To(Equal(ccintf.CCID{Name: "the-name", Version: "the-version"}))
+
+	ec, exitErr := exited.ArgsForCall(0)
+	gt.Expect(ec).To(Equal(99))
+	gt.Expect(exitErr).To(MatchError("boing-boing"))
+}