FAB-16158 Create streaming package retriever

This CR re-implements the package parsing in a streaming fashion, rather
than relying on pulling the whole code package into RAM as the previous
implementation did.  Note, the old implementation still exists, and is
still used by the non-streaming code, notably the lifecycle SCC.  We
could probably transition that code onto this interface.

Change-Id: Ib8216c57570b2f662359d4818a6614d53c4a2347
Signed-off-by: Jason Yellick <jyellick@us.ibm.com>

Author: Jason Yellick

core/chaincode/persistence/chaincode_package.go

@@ -13,8 +13,12 @@ import (
 	"encoding/json"
 	"io"
 	"io/ioutil"
+	"os"
+	"path/filepath"
 	"regexp"
 
+	"github.com/hyperledger/fabric/core/chaincode/persistence/intf" // yuck
+
 	"github.com/pkg/errors"
 )
 
@@ -41,6 +45,105 @@ const (
 	ChaincodePackageMetadataFile = "Chaincode-Package-Metadata.json"
 )
 
+type ChaincodePackageLocator struct {
+	ChaincodeDir string
+}
+
+func (cpl *ChaincodePackageLocator) ChaincodePackageStreamer(packageID persistence.PackageID) *ChaincodePackageStreamer {
+	return &ChaincodePackageStreamer{
+		PackagePath: filepath.Join(cpl.ChaincodeDir, packageID.String()+".bin"),
+	}
+}
+
+type ChaincodePackageStreamer struct {
+	PackagePath string
+}
+
+func (cps *ChaincodePackageStreamer) Metadata() (*ChaincodePackageMetadata, error) {
+	tarFileStream, err := cps.File(PreferredChaincodePackageMetadataFile)
+	if err != nil {
+		return nil, errors.WithMessage(err, "could not get metadata file")
+	}
+
+	defer tarFileStream.Close()
+
+	metadata := &ChaincodePackageMetadata{}
+	err = json.NewDecoder(tarFileStream).Decode(metadata)
+	if err != nil {
+		return nil, errors.WithMessage(err, "could not parse metadata file")
+	}
+
+	return metadata, nil
+}
+
+func (cps *ChaincodePackageStreamer) Code() (*TarFileStream, error) {
+	tarFileStream, err := cps.File(CodePackageFile)
+	if err != nil {
+		return nil, errors.WithMessage(err, "could not get code package")
+	}
+
+	return tarFileStream, nil
+}
+
+func (cps *ChaincodePackageStreamer) File(name string) (tarFileStream *TarFileStream, err error) {
+	file, err := os.Open(cps.PackagePath)
+	if err != nil {
+		return nil, errors.WithMessagef(err, "could not open chaincode package at '%s'", cps.PackagePath)
+	}
+
+	defer func() {
+		if err != nil {
+			file.Close()
+		}
+	}()
+
+	gzReader, err := gzip.NewReader(file)
+	if err != nil {
+		return nil, errors.Wrapf(err, "error reading as gzip stream")
+	}
+
+	tarReader := tar.NewReader(gzReader)
+
+	for {
+		header, err := tarReader.Next()
+		if err == io.EOF {
+			break
+		}
+
+		if err != nil {
+			return nil, errors.Wrapf(err, "error inspecting next tar header")
+		}
+
+		if header.Name != name {
+			continue
+		}
+
+		if header.Typeflag != tar.TypeReg {
+			return nil, errors.Errorf("tar entry %s is not a regular file, type %v", header.Name, header.Typeflag)
+		}
+
+		return &TarFileStream{
+			TarFile:    tarReader,
+			FileStream: file,
+		}, nil
+	}
+
+	return nil, errors.Errorf("did not find file '%s' in package", name)
+}
+
+type TarFileStream struct {
+	TarFile    io.Reader
+	FileStream io.Closer
+}
+
+func (tfs *TarFileStream) Read(p []byte) (int, error) {
+	return tfs.TarFile.Read(p)
+}
+
+func (tfs *TarFileStream) Close() error {
+	return tfs.FileStream.Close()
+}
+
 // ChaincodePackage represents the un-tar-ed format of the chaincode package.
 type ChaincodePackage struct {
 	Metadata    *ChaincodePackageMetadata

core/chaincode/persistence/chaincode_package_test.go

@@ -162,3 +162,147 @@ var _ = Describe("ChaincodePackageParser", func() {
 		})
 	})
 })
+
+var _ = Describe("ChaincodePackageLocator", func() {
+	var (
+		locator *persistence.ChaincodePackageLocator
+	)
+
+	BeforeEach(func() {
+		locator = &persistence.ChaincodePackageLocator{
+			ChaincodeDir: "/fake-dir",
+		}
+	})
+
+	Describe("ChaincodePackageStreamer", func() {
+		It("creates a ChaincodePackageStreamer for the given packageID", func() {
+			streamer := locator.ChaincodePackageStreamer("test-package")
+			Expect(streamer).To(Equal(&persistence.ChaincodePackageStreamer{
+				PackagePath: "/fake-dir/test-package.bin",
+			}))
+		})
+	})
+})
+
+var _ = Describe("ChaincodePackageStreamer", func() {
+	var (
+		streamer *persistence.ChaincodePackageStreamer
+	)
+
+	BeforeEach(func() {
+		streamer = &persistence.ChaincodePackageStreamer{
+			PackagePath: "testdata/good-package.tar.gz",
+		}
+	})
+
+	Describe("Metadata", func() {
+		It("reads the metadata from the package", func() {
+			md, err := streamer.Metadata()
+			Expect(err).NotTo(HaveOccurred())
+			Expect(md).To(Equal(&persistence.ChaincodePackageMetadata{
+				Type:  "Fake-Type",
+				Path:  "Fake-Path",
+				Label: "Real-Label",
+			}))
+		})
+
+		Context("when the metadata file cannot be found", func() {
+			BeforeEach(func() {
+				streamer.PackagePath = "testdata/missing-metadata.tar.gz"
+			})
+
+			It("wraps and returns the error", func() {
+				_, err := streamer.Metadata()
+				Expect(err).To(MatchError("could not get metadata file: did not find file 'metadata.json' in package"))
+			})
+		})
+
+		Context("when the metadata file cannot be parsed", func() {
+			BeforeEach(func() {
+				streamer.PackagePath = "testdata/bad-metadata.tar.gz"
+			})
+
+			It("wraps and returns the error", func() {
+				_, err := streamer.Metadata()
+				Expect(err).To(MatchError("could not parse metadata file: invalid character '\\n' in string literal"))
+			})
+		})
+	})
+
+	Describe("Code", func() {
+		It("reads a file from the package", func() {
+			code, err := streamer.Code()
+			Expect(err).NotTo(HaveOccurred())
+			codeBytes, err := ioutil.ReadAll(code)
+			code.Close()
+			Expect(err).NotTo(HaveOccurred())
+			Expect(codeBytes).To(Equal([]byte("package")))
+		})
+
+		Context("when the file cannot be found because the code is not a regular file", func() {
+			BeforeEach(func() {
+				streamer.PackagePath = "testdata/missing-codepackage.tar.gz"
+			})
+
+			It("wraps and returns the error", func() {
+				_, err := streamer.Code()
+				Expect(err).To(MatchError("could not get code package: did not find file 'code.tar.gz' in package"))
+			})
+		})
+	})
+
+	Describe("File", func() {
+		It("reads a file from the package", func() {
+			code, err := streamer.File("code.tar.gz")
+			Expect(err).NotTo(HaveOccurred())
+			codeBytes, err := ioutil.ReadAll(code)
+			code.Close()
+			Expect(err).NotTo(HaveOccurred())
+			Expect(codeBytes).To(Equal([]byte("package")))
+		})
+
+		Context("when the file is not a regular file", func() {
+			BeforeEach(func() {
+				streamer.PackagePath = "testdata/non-regular-file.tar.gz"
+			})
+
+			It("wraps and returns the error", func() {
+				_, err := streamer.File("code.tar.gz")
+				Expect(err).To(MatchError("tar entry code.tar.gz is not a regular file, type 50"))
+			})
+		})
+
+		Context("when the code cannot be found because the archive is corrupt", func() {
+			BeforeEach(func() {
+				streamer.PackagePath = "testdata/bad-archive.tar.gz"
+			})
+
+			It("wraps and returns the error", func() {
+				_, err := streamer.File("code.tar.gz")
+				Expect(err).To(MatchError("could not open chaincode package at 'testdata/bad-archive.tar.gz': open testdata/bad-archive.tar.gz: no such file or directory"))
+			})
+		})
+
+		Context("when the code cannot be found because the header is corrupt", func() {
+			BeforeEach(func() {
+				streamer.PackagePath = "testdata/corrupted-header.tar.gz"
+			})
+
+			It("wraps and returns the error", func() {
+				_, err := streamer.File("code.tar.gz")
+				Expect(err).To(MatchError("error inspecting next tar header: flate: corrupt input before offset 86"))
+			})
+		})
+
+		Context("when the code cannot be found because the gzip is corrupt", func() {
+			BeforeEach(func() {
+				streamer.PackagePath = "testdata/corrupted-gzip.tar.gz"
+			})
+
+			It("wraps and returns the error", func() {
+				_, err := streamer.File("code.tar.gz")
+				Expect(err).To(MatchError("error reading as gzip stream: unexpected EOF"))
+			})
+		})
+	})
+})

core/chaincode/persistence/testdata/corrupted-gzip.tar.gz

@@ -0,0 +1 @@
+notagzip