Improve performance of downloading objects and add cache to hypervisor.

cmd/hypervisor/main.go

@@ -36,7 +36,8 @@ var (
 		"Port number of image server")
 	networkBootImage = flag.String("networkBootImage", "pxelinux.0",
 		"Name of boot image passed via DHCP option")
-	portNum = flag.Uint("portNum", constants.HypervisorPortNumber,
+	objectCacheSize = flagutil.Size(10 << 30)
+	portNum         = flag.Uint("portNum", constants.HypervisorPortNumber,
 		"Port number to allocate and listen on for HTTP/RPC")
 	showVGA  = flag.Bool("showVGA", false, "If true, show VGA console")
 	stateDir = flag.String("stateDir", "/var/lib/hypervisor",
@@ -51,6 +52,8 @@ var (
 )
 
 func init() {
+	flag.Var(&objectCacheSize, "objectCacheSize",
+		"maximum size of object cache")
 	flag.Var(&volumeDirectories, "volumeDirectories",
 		"Comma separated list of volume directories. If empty, scan for space")
 }
@@ -120,6 +123,7 @@ func main() {
 		ImageServerAddress: imageServerAddress,
 		DhcpServer:         dhcpServer,
 		Logger:             logger,
+		ObjectCacheBytes:   uint64(objectCacheSize),
 		ShowVgaConsole:     *showVGA,
 		StateDir:           *stateDir,
 		Username:           *username,

cmd/imagetool/main.go

@@ -96,21 +96,22 @@ func printUsage() {
 	fmt.Fprintln(os.Stderr, "           i: name of an image on the imageserver")
 	fmt.Fprintln(os.Stderr, "           l: name of file containing an Image")
 	fmt.Fprintln(os.Stderr, "           s: name of sub to poll")
-	fmt.Fprintln(os.Stderr, "  estimate-usage    name")
-	fmt.Fprintln(os.Stderr, "  find-latest-image directory")
-	fmt.Fprintln(os.Stderr, "  get               name directory")
+	fmt.Fprintln(os.Stderr, "  estimate-usage      name")
+	fmt.Fprintln(os.Stderr, "  find-latest-image   directory")
+	fmt.Fprintln(os.Stderr, "  get                 name directory")
 	fmt.Fprintln(os.Stderr, "  list")
 	fmt.Fprintln(os.Stderr, "  listdirs")
 	fmt.Fprintln(os.Stderr, "  listunrefobj")
-	fmt.Fprintln(os.Stderr, "  list-latest-image directory")
-	fmt.Fprintln(os.Stderr, "  make-raw-image    name rawfile")
-	fmt.Fprintln(os.Stderr, "  match-triggers    name triggers-file")
-	fmt.Fprintln(os.Stderr, "  merge-filters     filter-file...")
-	fmt.Fprintln(os.Stderr, "  merge-triggers    triggers-file...")
-	fmt.Fprintln(os.Stderr, "  mkdir             name")
-	fmt.Fprintln(os.Stderr, "  show              name")
+	fmt.Fprintln(os.Stderr, "  list-latest-image   directory")
+	fmt.Fprintln(os.Stderr, "  make-raw-image      name rawfile")
+	fmt.Fprintln(os.Stderr, "  match-triggers      name triggers-file")
+	fmt.Fprintln(os.Stderr, "  merge-filters       filter-file...")
+	fmt.Fprintln(os.Stderr, "  merge-triggers      triggers-file...")
+	fmt.Fprintln(os.Stderr, "  mkdir               name")
+	fmt.Fprintln(os.Stderr, "  show                name")
 	fmt.Fprintln(os.Stderr, "  showunrefobj")
-	fmt.Fprintln(os.Stderr, "  tar               name [file]")
+	fmt.Fprintln(os.Stderr, "  tar                 name [file]")
+	fmt.Fprintln(os.Stderr, "  test-download-speed name")
 	fmt.Fprintln(os.Stderr, "Fields:")
 	fmt.Fprintln(os.Stderr, "  m: mode")
 	fmt.Fprintln(os.Stderr, "  l: number of hardlinks")
@@ -159,6 +160,7 @@ var subcommands = []subcommand{
 	{"show", 1, 1, showImageSubcommand},
 	{"showunrefobj", 0, 0, showUnreferencedObjectsSubcommand},
 	{"tar", 1, 2, tarImageSubcommand},
+	{"test-download-speed", 1, 1, testDownloadSpeedSubcommand},
 }
 
 var imageSrpcClient *srpc.Client

cmd/imagetool/testDownloadSpeed.go

@@ -0,0 +1,94 @@
+package main
+
+import (
+	"errors"
+	"fmt"
+	"io"
+	"io/ioutil"
+	"os"
+	"time"
+
+	imgclient "github.com/Symantec/Dominator/imageserver/client"
+	"github.com/Symantec/Dominator/lib/filesystem"
+	"github.com/Symantec/Dominator/lib/format"
+	"github.com/Symantec/Dominator/lib/hash"
+	"github.com/Symantec/Dominator/lib/log"
+	"github.com/Symantec/Dominator/lib/objectserver"
+)
+
+func testDownloadSpeedSubcommand(args []string) {
+	if err := testDownloadSpeed(args[0], logger); err != nil {
+		fmt.Fprintf(os.Stderr, "Error testing download speed: %s\n", err)
+		os.Exit(1)
+	}
+	os.Exit(0)
+}
+
+func testDownloadSpeed(imageName string, logger log.Logger) error {
+	imageSClient, objClient := getClients()
+	isDir, err := imgclient.CheckDirectory(imageSClient, imageName)
+	if err != nil {
+		return err
+	}
+	if isDir {
+		name, err := imgclient.FindLatestImage(imageSClient, imageName, false)
+		if err != nil {
+			return err
+		} else {
+			imageName = name
+		}
+	}
+	startTime := time.Now()
+	img, err := imgclient.GetImageWithTimeout(imageSClient, imageName, *timeout)
+	if err != nil {
+		return err
+	}
+	if img == nil {
+		return errors.New(imageName + ": not found")
+	}
+	finishedTime := time.Now()
+	logger.Printf("downloaded image metadata in %s\n",
+		format.Duration(finishedTime.Sub(startTime)))
+	hashes := make([]hash.Hash, 0, len(img.FileSystem.InodeTable))
+	for _, inode := range img.FileSystem.InodeTable {
+		if inode, ok := inode.(*filesystem.RegularInode); ok {
+			if inode.Size > 0 {
+				hashes = append(hashes, inode.Hash)
+			}
+		}
+	}
+	startTime = time.Now()
+	objectsReader, err := objClient.GetObjects(hashes)
+	if err != nil {
+		return err
+	}
+	defer objectsReader.Close()
+	var totalBytes uint64
+	buffer := make([]byte, 32<<10)
+	for range hashes {
+		if length, err := readOneObject(objectsReader, buffer); err != nil {
+			return err
+		} else {
+			totalBytes += length
+		}
+	}
+	finishedTime = time.Now()
+	downloadTime := finishedTime.Sub(startTime)
+	downloadSpeed := float64(totalBytes) / downloadTime.Seconds()
+	logger.Printf("downloaded %s (%d objects) in %s (%s/s)\n",
+		format.FormatBytes(totalBytes), len(hashes),
+		format.Duration(downloadTime),
+		format.FormatBytes(uint64(downloadSpeed)))
+	return nil
+}
+
+func readOneObject(objectsReader objectserver.ObjectsReader,
+	buffer []byte) (uint64, error) {
+	_, reader, err := objectsReader.NextObject()
+	if err != nil {
+		return 0, err
+	}
+	defer reader.Close()
+	nCopied, err := io.CopyBuffer(ioutil.Discard, reader, buffer)
+	return uint64(nCopied), err
+}

hypervisor/manager/api.go

@@ -7,6 +7,7 @@ import (
 	"time"
 
 	"github.com/Symantec/Dominator/lib/log"
+	"github.com/Symantec/Dominator/lib/objectserver/cachingreader"
 	"github.com/Symantec/Dominator/lib/srpc"
 	"github.com/Symantec/Dominator/lib/tags"
 	proto "github.com/Symantec/Dominator/proto/hypervisor"
@@ -37,6 +38,7 @@ type Manager struct {
 	addressPool       addressPoolType
 	healthStatus      string
 	notifiers         map[<-chan proto.Update]chan<- proto.Update
+	objectCache       *cachingreader.ObjectServer
 	ownerGroups       map[string]struct{}
 	ownerUsers        map[string]struct{}
 	subnets           map[string]proto.Subnet // Key: Subnet ID.
@@ -48,6 +50,7 @@ type StartOptions struct {
 	DhcpServer         DhcpServer
 	ImageServerAddress string
 	Logger             log.DebugLogger
+	ObjectCacheBytes   uint64
 	ShowVgaConsole     bool
 	StateDir           string
 	Username           string

hypervisor/manager/html.go

@@ -61,6 +61,11 @@ func (m *Manager) writeHtml(writer io.Writer) {
 		"Number of subnets: <a href=\"listSubnets\">%d</a><br>\n", numSubnets)
 	fmt.Fprintf(writer, "Volume directories: %s<br>\n",
 		strings.Join(m.volumeDirectories, " "))
+	if m.objectCache == nil {
+		fmt.Fprintln(writer, "No object cache<br>")
+	} else {
+		m.objectCache.WriteHtml(writer)
+	}
 }
 
 func writeCountLinks(writer io.Writer, text, path string, count uint) {

hypervisor/manager/start.go

@@ -5,7 +5,7 @@ import (
 	"crypto/rand"
 	"errors"
 	"os"
-	"path"
+	"path/filepath"
 	"runtime"
 	"strings"
 	"syscall"
@@ -15,6 +15,7 @@ import (
 	"github.com/Symantec/Dominator/lib/json"
 	"github.com/Symantec/Dominator/lib/log/prefixlogger"
 	"github.com/Symantec/Dominator/lib/meminfo"
+	"github.com/Symantec/Dominator/lib/objectserver/cachingreader"
 	"github.com/Symantec/Dominator/lib/rpcclientpool"
 	proto "github.com/Symantec/Dominator/proto/hypervisor"
 	"github.com/Symantec/tricorder/go/tricorder/messages"
@@ -38,8 +39,8 @@ func newManager(startOptions StartOptions) (*Manager, error) {
 	if _, err := rand.Read(importCookie); err != nil {
 		return nil, err
 	}
-	err = fsutil.CopyToFile(path.Join(startOptions.StateDir, "import-cookie"),
-		privateFilePerms, bytes.NewReader(importCookie), 0)
+	err = fsutil.CopyToFile(filepath.Join(startOptions.StateDir,
+		"import-cookie"), privateFilePerms, bytes.NewReader(importCookie), 0)
 	if err != nil {
 		return nil, err
 	}
@@ -59,7 +60,7 @@ func newManager(startOptions StartOptions) (*Manager, error) {
 	if err := manager.loadAddressPool(); err != nil {
 		return nil, err
 	}
-	dirname := path.Join(manager.StateDir, "VMs")
+	dirname := filepath.Join(manager.StateDir, "VMs")
 	dir, err := os.Open(dirname)
 	if err != nil {
 		if os.IsNotExist(err) {
@@ -80,8 +81,8 @@ func newManager(startOptions StartOptions) (*Manager, error) {
 			"error reading directory: " + dirname + ": " + err.Error())
 	}
 	for _, ipAddr := range names {
-		vmDirname := path.Join(dirname, ipAddr)
-		filename := path.Join(vmDirname, "info.json")
+		vmDirname := filepath.Join(dirname, ipAddr)
+		filename := filepath.Join(vmDirname, "info.json")
 		var vmInfo vmInfoType
 		if err := json.ReadFromFile(filename, &vmInfo); err != nil {
 			return nil, err
@@ -134,6 +135,20 @@ func newManager(startOptions StartOptions) (*Manager, error) {
 			return nil, err
 		}
 	}
+	if startOptions.ObjectCacheBytes >= 1<<20 {
+		dirname := filepath.Join(filepath.Dir(manager.volumeDirectories[0]),
+			"objectcache")
+		if err := os.MkdirAll(dirname, dirPerms); err != nil {
+			return nil, err
+		}
+		objSrv, err := cachingreader.NewObjectServer(dirname,
+			startOptions.ObjectCacheBytes, startOptions.ImageServerAddress,
+			startOptions.Logger)
+		if err != nil {
+			return nil, err
+		}
+		manager.objectCache = objSrv
+	}
 	go manager.loopCheckHealthStatus()
 	return manager, nil
 }