kubernetes
/
minikube
mirror of https://github.com/kubernetes/minikube.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

Refactor CommandRunner.Copy functions for consistency

pull/7559/head
Thomas Stromberg 2020-04-09 13:24:14 -07:00
parent 7fd834e62b
commit bf784d8bf3
10 changed files with 227 additions and 157 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
pkg/addons/addons.go
View File

@ -197,7 +197,7 @@ func enableOrDisableAddonInternal(cc *config.ClusterConfig, addon *assets.Addon,
if addon.IsTemplate() {
f, err = addon.Evaluate(data)
if err != nil {
return errors.Wrapf(err, "evaluate bundled addon %s asset", addon.GetAssetName())
return errors.Wrapf(err, "evaluate bundled addon %s asset", addon.GetSourcePath())
}
} else {

2
pkg/gvisor/enable.go
View File

@ -181,7 +181,7 @@ func copyAssetToDest(targetName, dest string) error {
log.Printf("%s asset path: %s", targetName, src)
contents, err := ioutil.ReadFile(src)
if err != nil {
return errors.Wrapf(err, "getting contents of %s", asset.GetAssetName())
return errors.Wrapf(err, "getting contents of %s", asset.GetSourcePath())
}
if _, err := os.Stat(dest); err == nil {
if err := os.Remove(dest); err != nil {

34
pkg/minikube/assets/vm_assets.go
View File

@ -29,11 +29,15 @@ import (
"github.com/pkg/errors"
)
// MemorySource is the source name used for in-memory copies
const MemorySource = "memory"
// CopyableFile is something that can be copied
type CopyableFile interface {
io.Reader
GetLength() int
GetAssetName() string
GetSourcePath() string
GetTargetDir() string
GetTargetName() string
GetPermissions() string
@ -43,15 +47,16 @@ type CopyableFile interface {
// BaseAsset is the base asset class
type BaseAsset struct {
AssetName string
SourcePath string
TargetDir string
TargetName string
Permissions string
Source string
}
// GetAssetName returns asset name
func (b *BaseAsset) GetAssetName() string {
return b.AssetName
// GetSourcePath returns asset name
func (b *BaseAsset) GetSourcePath() string {
return b.SourcePath
}
// GetTargetDir returns target dir
@ -99,7 +104,7 @@ func NewFileAsset(src, targetDir, targetName, permissions string) (*FileAsset, e
r := io.NewSectionReader(f, 0, info.Size())
return &FileAsset{
BaseAsset: BaseAsset{
AssetName: src,
SourcePath: src,
TargetDir: targetDir,
TargetName: targetName,
Permissions: permissions,
@ -110,7 +115,7 @@ func NewFileAsset(src, targetDir, targetName, permissions string) (*FileAsset, e
// GetLength returns the file length, or 0 (on error)
func (f *FileAsset) GetLength() (flen int) {
fi, err := os.Stat(f.AssetName)
fi, err := os.Stat(f.SourcePath)
if err != nil {
return 0
}
@ -119,7 +124,7 @@ func (f *FileAsset) GetLength() (flen int) {
// GetModTime returns modification time of the file
func (f *FileAsset) GetModTime() (time.Time, error) {
fi, err := os.Stat(f.AssetName)
fi, err := os.Stat(f.SourcePath)
if err != nil {
return time.Time{}, err
}
@ -168,6 +173,7 @@ func NewMemoryAsset(d []byte, targetDir, targetName, permissions string) *Memory
TargetDir: targetDir,
TargetName: targetName,
Permissions: permissions,
SourcePath: MemorySource,
},
reader: bytes.NewReader(d),
length: len(d),
@ -195,7 +201,7 @@ func MustBinAsset(name, targetDir, targetName, permissions string, isTemplate bo
func NewBinAsset(name, targetDir, targetName, permissions string, isTemplate bool) (*BinAsset, error) {
m := &BinAsset{
BaseAsset: BaseAsset{
AssetName: name,
SourcePath: name,
TargetDir: targetDir,
TargetName: targetName,
Permissions: permissions,
@ -218,13 +224,13 @@ func defaultValue(defValue string, val interface{}) string {
}
func (m *BinAsset) loadData(isTemplate bool) error {
contents, err := Asset(m.AssetName)
contents, err := Asset(m.SourcePath)
if err != nil {
return err
}
if isTemplate {
tpl, err := template.New(m.AssetName).Funcs(template.FuncMap{"default": defaultValue}).Parse(string(contents))
tpl, err := template.New(m.SourcePath).Funcs(template.FuncMap{"default": defaultValue}).Parse(string(contents))
if err != nil {
return err
}
@ -234,9 +240,9 @@ func (m *BinAsset) loadData(isTemplate bool) error {
m.length = len(contents)
m.reader = bytes.NewReader(contents)
glog.V(1).Infof("Created asset %s with %d bytes", m.AssetName, m.length)
glog.V(1).Infof("Created asset %s with %d bytes", m.SourcePath, m.length)
if m.length == 0 {
return fmt.Errorf("%s is an empty asset", m.AssetName)
return fmt.Errorf("%s is an empty asset", m.SourcePath)
}
return nil
}
@ -249,7 +255,7 @@ func (m *BinAsset) IsTemplate() bool {
// Evaluate evaluates the template to a new asset
func (m *BinAsset) Evaluate(data interface{}) (*MemoryAsset, error) {
if !m.IsTemplate() {
return nil, errors.Errorf("the asset %s is not a template", m.AssetName)
return nil, errors.Errorf("the asset %s is not a template", m.SourcePath)
}

3
pkg/minikube/bootstrapper/certs.go
View File

@ -117,9 +117,8 @@ func SetupCerts(cmd command.Runner, k8s config.KubernetesConfig, n config.Node)
}
for _, f := range copyableFiles {
glog.Infof("copying: %s/%s", f.GetTargetDir(), f.GetTargetName())
if err := cmd.Copy(f); err != nil {
return nil, errors.Wrapf(err, "Copy %s", f.GetAssetName())
return nil, errors.Wrapf(err, "Copy %s", f.GetSourcePath())
}
}

109
pkg/minikube/command/command_runner.go
View File

@ -17,12 +17,17 @@ limitations under the License.
package command
import (
"bufio"
"bytes"
"fmt"
"io"
"os"
"os/exec"
"path"
"strconv"
"strings"
"time"
"github.com/pkg/errors"
"k8s.io/minikube/pkg/minikube/assets"
)
@ -55,10 +60,6 @@ type Runner interface {
Remove(assets.CopyableFile) error
}
func getDeleteFileCommand(f assets.CopyableFile) string {
return fmt.Sprintf("sudo rm %s", path.Join(f.GetTargetDir(), f.GetTargetName()))
}
// Command returns a human readable command string that does not induce eye fatigue
func (rr RunResult) Command() string {
var sb strings.Builder
@ -84,3 +85,101 @@ func (rr RunResult) Output() string {
}
return sb.String()
}
// teePrefix copies bytes from a reader to writer, logging each new line.
func teePrefix(prefix string, r io.Reader, w io.Writer, logger func(format string, args ...interface{})) error {
scanner := bufio.NewScanner(r)
scanner.Split(bufio.ScanBytes)
var line bytes.Buffer
for scanner.Scan() {
b := scanner.Bytes()
if _, err := w.Write(b); err != nil {
return err
}
if bytes.IndexAny(b, "\r\n") == 0 {
if line.Len() > 0 {
logger("%s%s", prefix, line.String())
line.Reset()
}
continue
}
line.Write(b)
}
// Catch trailing output in case stream does not end with a newline
if line.Len() > 0 {
logger("%s%s", prefix, line.String())
}
return nil
}
// fileExists checks that the same file exists on the other end
func fileExists(r Runner, f assets.CopyableFile, dst string) (bool, error) {
// It's too difficult to tell if the file exists with the exact contents
if f.GetSourcePath() == assets.MemorySource {
return false, nil
}
// get file size and modtime of the source
srcSize := f.GetLength()
srcModTime, err := f.GetModTime()
if err != nil {
return false, err
}
if srcModTime.IsZero() {
return false, nil
}
// get file size and modtime of the destination
rr, err := r.RunCmd(exec.Command("stat", "-c", "%s %y", dst))
if err != nil {
if rr.ExitCode == 1 {
return false, nil
}
// avoid the noise because ssh doesn't propagate the exit code
if strings.HasSuffix(err.Error(), "status 1") {
return false, nil
}
return false, err
}
stdout := strings.TrimSpace(rr.Stdout.String())
outputs := strings.SplitN(stdout, " ", 2)
dstSize, err := strconv.Atoi(outputs[0])
if err != nil {
return false, err
}
dstModTime, err := time.Parse(layout, outputs[1])
if err != nil {
return false, err
}
if srcSize != dstSize {
return false, errors.New("source file and destination file are different sizes")
}
return srcModTime.Equal(dstModTime), nil
}
// writeFile is like ioutil.WriteFile, but does not require reading file into memory
func writeFile(dst string, f assets.CopyableFile, perms os.FileMode) error {
w, err := os.OpenFile(dst, os.O_WRONLY|os.O_CREATE, perms)
if err != nil {
return errors.Wrap(err, "create")
}
defer w.Close()
r := f.(io.Reader)
n, err := io.Copy(w, r)
if err != nil {
return errors.Wrap(err, "copy")
}
if n != int64(f.GetLength()) {
return fmt.Errorf("%s: expected to write %d bytes, but wrote %d instead", dst, f.GetLength(), n)
}
return w.Close()
}

34
pkg/minikube/command/exec_runner.go
View File

@ -86,35 +86,31 @@ func (*execRunner) RunCmd(cmd *exec.Cmd) (*RunResult, error) {
// Copy copies a file and its permissions
func (*execRunner) Copy(f assets.CopyableFile) error {
targetPath := path.Join(f.GetTargetDir(), f.GetTargetName())
if _, err := os.Stat(targetPath); err == nil {
if err := os.Remove(targetPath); err != nil {
return errors.Wrapf(err, "error removing file %s", targetPath)
dst := path.Join(f.GetTargetDir(), f.GetTargetName())
if _, err := os.Stat(dst); err == nil {
glog.Infof("found %s, removing ...", dst)
if err := os.Remove(dst); err != nil {
return errors.Wrapf(err, "error removing file %s", dst)
}
}
src := f.GetSourcePath()
glog.Infof("cp: %s --> %s (%d bytes)", src, dst, f.GetLength())
if f.GetLength() == 0 {
glog.Warningf("0 byte asset: %+v", f)
}
target, err := os.Create(targetPath)
if err != nil {
return errors.Wrapf(err, "error creating file at %s", targetPath)
}
perms, err := strconv.ParseInt(f.GetPermissions(), 8, 0)
if err != nil {
return errors.Wrapf(err, "error converting permissions %s to integer", f.GetPermissions())
}
if err := os.Chmod(targetPath, os.FileMode(perms)); err != nil {
return errors.Wrapf(err, "error changing file permissions for %s", targetPath)
}
if _, err = io.Copy(target, f); err != nil {
return errors.Wrapf(err, `error copying file %s to target location:
do you have the correct permissions?`,
targetPath)
}
return target.Close()
return writeFile(dst, f, os.FileMode(perms))
}
// Remove removes a file
func (*execRunner) Remove(f assets.CopyableFile) error {
targetPath := filepath.Join(f.GetTargetDir(), f.GetTargetName())
return os.Remove(targetPath)
dst := filepath.Join(f.GetTargetDir(), f.GetTargetName())
glog.Infof("rm: %s", dst)
return os.Remove(dst)
}

4
pkg/minikube/command/fake_runner.go
View File

@ -97,13 +97,13 @@ func (f *FakeCommandRunner) Copy(file assets.CopyableFile) error {
if err != nil {
return errors.Wrapf(err, "error reading file: %+v", file)
}
f.fileMap.Store(file.GetAssetName(), b.String())
f.fileMap.Store(file.GetSourcePath(), b.String())
return nil
}
// Remove removes the filename, file contents key value pair from the stored map
func (f *FakeCommandRunner) Remove(file assets.CopyableFile) error {
f.fileMap.Delete(file.GetAssetName())
f.fileMap.Delete(file.GetSourcePath())
return nil
}

87
pkg/minikube/command/kic_runner.go
View File

@ -128,44 +128,73 @@ func (k *kicRunner) RunCmd(cmd *exec.Cmd) (*RunResult, error) {
// Copy copies a file and its permissions
func (k *kicRunner) Copy(f assets.CopyableFile) error {
src := f.GetAssetName()
if _, err := os.Stat(f.GetAssetName()); os.IsNotExist(err) {
fc := make([]byte, f.GetLength()) // Read asset file into a []byte
if _, err := f.Read(fc); err != nil {
return errors.Wrap(err, "can't copy non-existing file")
} // we have a MemoryAsset, will write to disk before copying
dst := path.Join(path.Join(f.GetTargetDir(), f.GetTargetName()))
tmpFile, err := ioutil.TempFile(os.TempDir(), "tmpf-memory-asset")
// For tiny files, it's cheaper to overwrite than check
if f.GetLength() > 4096 {
exists, err := fileExists(k, f, dst)
if err != nil {
return errors.Wrap(err, "creating temporary file")
glog.Infof("existence error for %s: %v", dst, err)
}
// clean up the temp file
defer os.Remove(tmpFile.Name())
if _, err = tmpFile.Write(fc); err != nil {
return errors.Wrap(err, "write to temporary file")
if exists {
glog.Infof("copy: skipping %s (exists)", dst)
return nil
}
}
// Close the file
if err := tmpFile.Close(); err != nil {
return errors.Wrap(err, "close temporary file")
}
src = tmpFile.Name()
src := f.GetSourcePath()
if f.GetLength() == 0 {
glog.Warningf("0 byte asset: %+v", f)
}
perms, err := strconv.ParseInt(f.GetPermissions(), 8, 0)
if err != nil {
return errors.Wrapf(err, "converting permissions %s to integer", f.GetPermissions())
return errors.Wrapf(err, "error converting permissions %s to integer", f.GetPermissions())
}
// Rely on cp -a to propagate permissions
if err := os.Chmod(src, os.FileMode(perms)); err != nil {
return errors.Wrapf(err, "chmod")
if src != assets.MemorySource {
// Take the fast path
fi, err := os.Stat(src)
if err == nil {
if fi.Mode() == os.FileMode(perms) {
glog.Infof("%s (direct): %s --> %s (%d bytes)", k.ociBin, src, dst, f.GetLength())
return k.copy(src, dst)
}
// If >1MB, avoid local copy
if fi.Size() > (1024 * 1024) {
glog.Infof("%s (chmod): %s --> %s (%d bytes)", k.ociBin, src, dst, f.GetLength())
if err := k.copy(src, dst); err != nil {
return err
}
return k.chmod(dst, f.GetPermissions())
}
}
}
dest := fmt.Sprintf("%s:%s", k.nameOrID, path.Join(f.GetTargetDir(), f.GetTargetName()))
glog.Infof("%s (temp): %s --> %s (%d bytes)", k.ociBin, src, dst, f.GetLength())
tf, err := ioutil.TempFile("", "tmpf-memory-asset")
if err != nil {
return errors.Wrap(err, "creating temporary file")
}
defer os.Remove(tf.Name())
if err := writeFile(tf.Name(), f, os.FileMode(perms)); err != nil {
return errors.Wrap(err, "write")
}
return k.copy(tf.Name(), dst)
}
func (k *kicRunner) copy(src string, dst string) error {
fullDest := fmt.Sprintf("%s:%s", k.nameOrID, dst)
if k.ociBin == oci.Podman {
return copyToPodman(src, dest)
return copyToPodman(src, fullDest)
}
return copyToDocker(src, dest)
return copyToDocker(src, fullDest)
}
func (k *kicRunner) chmod(dst string, perm string) error {
_, err := k.RunCmd(exec.Command("sudo", "chmod", perm, dst))
return err
}
// Podman cp command doesn't match docker and doesn't have -a
@ -185,11 +214,11 @@ func copyToDocker(src string, dest string) error {
// Remove removes a file
func (k *kicRunner) Remove(f assets.CopyableFile) error {
fp := path.Join(f.GetTargetDir(), f.GetTargetName())
if rr, err := k.RunCmd(exec.Command("sudo", "rm", fp)); err != nil {
return errors.Wrapf(err, "removing file %q output: %s", fp, rr.Output())
}
return nil
dst := path.Join(f.GetTargetDir(), f.GetTargetName())
glog.Infof("rm: %s", dst)
_, err := k.RunCmd(exec.Command("sudo", "rm", dst))
return err
}
// isTerminal returns true if the writer w is a terminal

107
pkg/minikube/command/ssh_runner.go
View File

@ -17,14 +17,11 @@ limitations under the License.
package command
import (
"bufio"
"bytes"
"fmt"
"io"
"os/exec"
"path"
"strconv"
"strings"
"sync"
"time"
@ -55,13 +52,16 @@ func NewSSHRunner(c *ssh.Client) *SSHRunner {
// Remove runs a command to delete a file on the remote.
func (s *SSHRunner) Remove(f assets.CopyableFile) error {
dst := path.Join(f.GetTargetDir(), f.GetTargetName())
glog.Infof("rm: %s", dst)
sess, err := s.c.NewSession()
if err != nil {
return errors.Wrap(err, "getting ssh session")
}
defer sess.Close()
cmd := getDeleteFileCommand(f)
return sess.Run(cmd)
return sess.Run(fmt.Sprintf("sudo rm %s", dst))
}
// teeSSH runs an SSH command, streaming stdout, stderr to logs
@ -150,14 +150,26 @@ func (s *SSHRunner) RunCmd(cmd *exec.Cmd) (*RunResult, error) {
// Copy copies a file to the remote over SSH.
func (s *SSHRunner) Copy(f assets.CopyableFile) error {
dst := path.Join(path.Join(f.GetTargetDir(), f.GetTargetName()))
exists, err := s.sameFileExists(f, dst)
if err != nil {
glog.Infof("Checked if %s exists, but got error: %v", dst, err)
// For small files, don't bother risking being wrong for no performance benefit
if f.GetLength() > 2048 {
exists, err := fileExists(s, f, dst)
if err != nil {
glog.Infof("existence check for %s: %v", dst, err)
}
if exists {
glog.Infof("copy: skipping %s (exists)", dst)
return nil
}
}
if exists {
glog.Infof("Skipping copying %s as it already exists", dst)
return nil
src := f.GetSourcePath()
glog.Infof("scp %s --> %s (%d bytes)", src, dst, f.GetLength())
if f.GetLength() == 0 {
glog.Warningf("0 byte asset: %+v", f)
}
sess, err := s.c.NewSession()
if err != nil {
return errors.Wrap(err, "NewSession")
@ -171,14 +183,13 @@ func (s *SSHRunner) Copy(f assets.CopyableFile) error {
// StdinPipe is closed. But let's use errgroup to make it explicit.
var g errgroup.Group
var copied int64
glog.Infof("Transferring %d bytes to %s", f.GetLength(), dst)
g.Go(func() error {
defer w.Close()
header := fmt.Sprintf("C%s %d %s\n", f.GetPermissions(), f.GetLength(), f.GetTargetName())
fmt.Fprint(w, header)
if f.GetLength() == 0 {
glog.Warningf("%s is a 0 byte asset!", f.GetTargetName())
glog.Warningf("asked to copy a 0 byte asset: %+v", f)
fmt.Fprint(w, "\x00")
return nil
}
@ -190,7 +201,6 @@ func (s *SSHRunner) Copy(f assets.CopyableFile) error {
if copied != int64(f.GetLength()) {
return fmt.Errorf("%s: expected to copy %d bytes, but copied %d instead", f.GetTargetName(), f.GetLength(), copied)
}
glog.Infof("%s: copied %d bytes", f.GetTargetName(), copied)
fmt.Fprint(w, "\x00")
return nil
})
@ -208,72 +218,3 @@ func (s *SSHRunner) Copy(f assets.CopyableFile) error {
}
return g.Wait()
}
func (s *SSHRunner) sameFileExists(f assets.CopyableFile, dst string) (bool, error) {
// get file size and modtime of the source
srcSize := f.GetLength()
srcModTime, err := f.GetModTime()
if err != nil {
return false, err
}
if srcModTime.IsZero() {
return false, nil
}
// get file size and modtime of the destination
sess, err := s.c.NewSession()
if err != nil {
return false, err
}
cmd := "stat -c \"%s %y\" " + dst
out, err := sess.CombinedOutput(cmd)
if err != nil {
return false, err
}
outputs := strings.SplitN(strings.Trim(string(out), "\n"), " ", 2)
dstSize, err := strconv.Atoi(outputs[0])
if err != nil {
return false, err
}
dstModTime, err := time.Parse(layout, outputs[1])
if err != nil {
return false, err
}
glog.Infof("found %s: %d bytes, modified at %s", dst, dstSize, dstModTime)
// compare sizes and modtimes
if srcSize != dstSize {
return false, errors.New("source file and destination file are different sizes")
}
return srcModTime.Equal(dstModTime), nil
}
// teePrefix copies bytes from a reader to writer, logging each new line.
func teePrefix(prefix string, r io.Reader, w io.Writer, logger func(format string, args ...interface{})) error {
scanner := bufio.NewScanner(r)
scanner.Split(bufio.ScanBytes)
var line bytes.Buffer
for scanner.Scan() {
b := scanner.Bytes()
if _, err := w.Write(b); err != nil {
return err
}
if bytes.IndexAny(b, "\r\n") == 0 {
if line.Len() > 0 {
logger("%s%s", prefix, line.String())
line.Reset()
}
continue
}
line.Write(b)
}
// Catch trailing output in case stream does not end with a newline
if line.Len() > 0 {
logger("%s%s", prefix, line.String())
}
return nil
}

2
pkg/minikube/machine/filesync_test.go
View File

@ -149,7 +149,7 @@ func TestAssetsFromDir(t *testing.T) {
got := make(map[string]string)
for _, actualFile := range actualFiles {
got[actualFile.GetAssetName()] = actualFile.GetTargetDir()
got[actualFile.GetSourcePath()] = actualFile.GetTargetDir()
}
if diff := cmp.Diff(want, got); diff != "" {
t.Errorf("files differ: (-want +got)\n%s", diff)