Вступление

Данная статья является третьей в цикле (1,2), посвященном изучению исходного кода Docker и прямым продолжением предыдущей статьи, в которой мы начали разбирать код первого публичного релиза Docker v0.1.0. В этой части будет рассмотрена реализация практически всех команд, а в конце, мы создадим образ и запустим докер контейнер на его основе. Для удобства я постарался разбить список команд на условные группы: работа с образами, работа с контейнерами, сетевой стек и т.д.

А теперь, как говорится, “without further ado”, приступим к изучению кода из файла commands.go начиная с команд для управления образами (images).

Управление образами

Import

Команда import позволяет импортировать образ файловой системы из tar архива, подаваемого на stdin, или же загрузить его по url:

CmdImport

func (srv *Server) CmdImport(stdin io.ReadCloser, stdout io.Writer, args ...string) error {
	cmd := rcli.Subcmd(stdout, "import", "[OPTIONS] URL|- [REPOSITORY [TAG]]", "Create a new filesystem image from the contents of a tarball")
	var archive io.Reader
	var resp *http.Response

	if err := cmd.Parse(args); err != nil {
		return nil
	}
	src := cmd.Arg(0)
	if src == "" {
		return errors.New("Not enough arguments")
	} else if src == "-" {
		archive = stdin
	} else {
		u, err := url.Parse(src)
		if err != nil {
			return err
		}
		if u.Scheme == "" {
			u.Scheme = "http"
			u.Host = src
			u.Path = ""
		}
		fmt.Fprintf(stdout, "Downloading from %s\n", u.String())
		// Download with curl (pretty progress bar)
		// If curl is not available, fallback to http.Get()
		resp, err = Download(u.String(), stdout)
		if err != nil {
			return err
		}
		archive = ProgressReader(resp.Body, int(resp.ContentLength), stdout)
	}
	img, err := srv.runtime.graph.Create(archive, nil, "Imported from "+src)
	if err != nil {
		return err
	}
	// Optionally register the image at REPO/TAG
	if repository := cmd.Arg(1); repository != "" {
		tag := cmd.Arg(2) // Repository will handle an empty tag properly
		if err := srv.runtime.repositories.Set(repository, tag, img.Id, true); err != nil {
			return err
		}
	}
	fmt.Fprintln(stdout, img.Id)
	return nil
}

После стандартного разбора аргументов функция определяет место откуда нужно импортировать образ: "-" означает stdin, в другом случае аргумент рассматривается, как url. Хелпер методы Download и ProgressReader для загрузки архива по http находятся в файле utils.go:

utils.go

// Request a given URL and return an io.Reader
func Download(url string, stderr io.Writer) (*http.Response, error) {
	var resp *http.Response
	var err error = nil
	if resp, err = http.Get(url); err != nil {
		return nil, err
	}
	if resp.StatusCode >= 400 {
		return nil, errors.New("Got HTTP status code >= 400: " + resp.Status)
	}
	return resp, nil
} 

type progressReader struct {
	reader        io.ReadCloser // Stream to read from
	output        io.Writer     // Where to send progress bar to
	read_total    int           // Expected stream length (bytes)
	read_progress int           // How much has been read so far (bytes)
	last_update   int           // How many bytes read at least update
}

func (r *progressReader) Read(p []byte) (n int, err error) {
	read, err := io.ReadCloser(r.reader).Read(p)
	r.read_progress += read

	// Only update progress for every 1% read
	update_every := int(0.01 * float64(r.read_total))
	if r.read_progress-r.last_update > update_every || r.read_progress == r.read_total {
		fmt.Fprintf(r.output, "%d/%d (%.0f%%)\r",
			r.read_progress,
			r.read_total,
			float64(r.read_progress)/float64(r.read_total)*100)
		r.last_update = r.read_progress
	}
	// Send newline when complete
	if err == io.EOF {
		fmt.Fprintf(r.output, "\n")
	}

	return read, err
}
func (r *progressReader) Close() error {
	return io.ReadCloser(r.reader).Close()
}
func ProgressReader(r io.ReadCloser, size int, output io.Writer) *progressReader {
	return &progressReader{r, output, size, 0, 0}
}

Далее управление переходит в функцию graph.Create из файла graph.go:

graph.Create

func (graph *Graph) Create(layerData Archive, container *Container, comment string) (*Image, error) {
	img := &Image{
		Id:      GenerateId(),
		Comment: comment,
		Created: time.Now(),
	}
	if container != nil {
		img.Parent = container.Image
		img.Container = container.Id
		img.ContainerConfig = *container.Config
	}
	if err := graph.Register(layerData, img); err != nil {
		return nil, err
	}
	return img, nil
}

Здесь генерируется уникальный идентификатор образа и инициализируется структура Image, которая далее вместе с данными архива передается в метод graph.Register. Если дополнительно передан и контейнер, то ссылка на его образ будет сохранена в поле img.Parent - это используется в команде Commit, создающей новый образ из текущего контейнера. Структура Image и функции для генерации Id на основе SHA256 находятся в файле image.go:

image.go

type Image struct {
	Id              string    `json:"id"`
	Parent          string    `json:"parent,omitempty"`
	Comment         string    `json:"comment,omitempty"`
	Created         time.Time `json:"created"`
	Container       string    `json:"container,omitempty"`
	ContainerConfig Config    `json:"container_config,omitempty"`
	graph           *Graph
}

func GenerateId() string {
	// FIXME: don't seed every time
	rand.Seed(time.Now().UTC().UnixNano())
	randomBytes := bytes.NewBuffer([]byte(fmt.Sprintf("%x", rand.Int())))
	id, _ := ComputeId(randomBytes) // can't fail
	return id
}

// ComputeId reads from `content` until EOF, then returns a SHA of what it read, as a string.
func ComputeId(content io.Reader) (string, error) {
	h := sha256.New()
	if _, err := io.Copy(h, content); err != nil {
		return "", err
	}
	return fmt.Sprintf("%x", h.Sum(nil)[:8]), nil
}

Далее взглянем на метод graph.Register:

graph.Register

func (graph *Graph) Register(layerData Archive, img *Image) error {
	if err := ValidateId(img.Id); err != nil {
		return err
	}
	// (This is a convenience to save time. Race conditions are taken care of by os.Rename)
	if graph.Exists(img.Id) {
		return fmt.Errorf("Image %s already exists", img.Id)
	}
	tmp, err := graph.Mktemp(img.Id)
	defer os.RemoveAll(tmp)
	if err != nil {
		return fmt.Errorf("Mktemp failed: %s", err)
	}
	if err := StoreImage(img, layerData, tmp); err != nil {
		return err
	}
	// Commit
	if err := os.Rename(tmp, graph.imageRoot(img.Id)); err != nil {
		return err
	}
	img.graph = graph
	return nil
}

После валидации id на наличие запрещенного символа ":" (так как он является разделителем для тега), создается временная папка для модификаций, а затем вызывается функция StoreImage, в которой происходит создание образа. По завершению временная папка переименовывается в img.Id:

StoreImage

func StoreImage(img *Image, layerData Archive, root string) error {
	// Check that root doesn't already exist
	if _, err := os.Stat(root); err == nil {
		return fmt.Errorf("Image %s already exists", img.Id)
	} else if !os.IsNotExist(err) {
		return err
	}
	// Store the layer
	layer := layerPath(root)
	if err := os.MkdirAll(layer, 0700); err != nil {
		return err
	}
	if err := Untar(layerData, layer); err != nil {
		return err
	}
	// Store the json ball
	jsonData, err := json.Marshal(img)
	if err != nil {
		return err
	}
	if err := ioutil.WriteFile(jsonPath(root), jsonData, 0600); err != nil {
		return err
	}
	return nil
}

func layerPath(root string) string {
	return path.Join(root, "layer")
}

func jsonPath(root string) string {
	return path.Join(root, "json")
}

В StoreImage создается директория layer, в которую помещается распакованный при помощи функции Untar архив файловой системы, после чего структура Image экспортируется в json и сохраняется в соседний файл, как метаданные для образа. Функции Tar и Untar для работы с архивами находятся в файле archive.go и представляют собой лишь удобные обертки над утилитой bsdtar:

archive.go

type Archive io.Reader

type Compression uint32

const (
	Uncompressed Compression = iota
	Bzip2
	Gzip
)

func (compression *Compression) Flag() string {
	switch *compression {
	case Bzip2:
		return "j"
	case Gzip:
		return "z"
	}
	return ""
}

func Tar(path string, compression Compression) (io.Reader, error) {
	cmd := exec.Command("bsdtar", "-f", "-", "-C", path, "-c"+compression.Flag(), ".")
	return CmdStream(cmd)
}

func Untar(archive io.Reader, path string) error {
	cmd := exec.Command("bsdtar", "-f", "-", "-C", path, "-x")
	cmd.Stdin = archive
	output, err := cmd.CombinedOutput()
	if err != nil {
		return errors.New(err.Error() + ": " + string(output))
	}
	return nil
}

func CmdStream(cmd *exec.Cmd) (io.Reader, error) {
	stdout, err := cmd.StdoutPipe()
	if err != nil {
		return nil, err
	}
	stderr, err := cmd.StderrPipe()
	if err != nil {
		return nil, err
	}
	pipeR, pipeW := io.Pipe()
	go func() {
		_, err := io.Copy(pipeW, stdout)
		if err != nil {
			pipeW.CloseWithError(err)
		}
		errText, e := ioutil.ReadAll(stderr)
		if e != nil {
			errText = []byte("(...couldn't fetch stderr: " + e.Error() + ")")
		}
		if err := cmd.Wait(); err != nil {
			// FIXME: can this block if stderr outputs more than the size of StderrPipe()'s buffer?
			pipeW.CloseWithError(errors.New(err.Error() + ": " + string(errText)))
		} else {
			pipeW.Close()
		}
	}()
	if err := cmd.Start(); err != nil {
		return nil, err
	}
	return pipeR, nil
}

Отметим, что функция import также может сохранять tag образа, принимая его опциональным параметром, но этот функционал мы рассмотрим позже, когда до него дойдет очередь.

Export

Команда export возвращает экспортированный архив файловой системы контейнера:

CmdExport

func (srv *Server) CmdExport(stdin io.ReadCloser, stdout io.Writer, args ...string) error {
	cmd := rcli.Subcmd(stdout,
		"export", "CONTAINER",
		"Export the contents of a filesystem as a tar archive")
	if err := cmd.Parse(args); err != nil {
		return nil
	}
	name := cmd.Arg(0)
	if container := srv.runtime.Get(name); container != nil {
		data, err := container.Export()
		if err != nil {
			return err
		}
		// Stream the entire contents of the container (basically a volatile snapshot)
		if _, err := io.Copy(stdout, data); err != nil {
			return err
		}
		return nil
	}
	return errors.New("No such container: " + name)
}

func (container *Container) Export() (Archive, error) {
	if err := container.EnsureMounted(); err != nil {
		return nil, err
	}
	return Tar(container.RootfsPath(), Uncompressed)
}

Функция по переданному имени получает контейнер и вызывает у него метод container.Export, который в свою очередь просто возвращает созданный архив, смонтированной директории Rootfs. Код функции Tar был приведен выше в файле archive.go.

Rmi

Удаляет переданный список образов, вызывая метод graph.Delete:

graph.Delete

func (srv *Server) CmdRmi(stdin io.ReadCloser, stdout io.Writer, args ...string) (err error) {
	cmd := rcli.Subcmd(stdout, "rmimage", "[OPTIONS] IMAGE", "Remove an image")
	if cmd.Parse(args) != nil || cmd.NArg() < 1 {
		cmd.Usage()
		return nil
	}
	for _, name := range cmd.Args() {
		if err := srv.runtime.graph.Delete(name); err != nil {
			return err
		}
	}
	return nil
}

func (graph *Graph) Delete(id string) error {
	garbage, err := graph.Garbage()
	if err != nil {
		return err
	}
	return os.Rename(graph.imageRoot(id), garbage.imageRoot(id))
}

func (graph *Graph) Garbage() (*Graph, error) {
	return NewGraph(path.Join(graph.Root, ":garbage:"))
}

В реальности, graph.Delete перемещает их в папку :garbage:, для возможности последующего восстановления, но данная функция здесь не используется.

Images

Возвращает таблицу со списком имеющихся образов:

CmdImages

func (srv *Server) CmdImages(stdin io.ReadCloser, stdout io.Writer, args ...string) error {
	cmd := rcli.Subcmd(stdout, "images", "[OPTIONS] [NAME]", "List images")
	//limit := cmd.Int("l", 0, "Only show the N most recent versions of each image")
	quiet := cmd.Bool("q", false, "only show numeric IDs")
	fl_a := cmd.Bool("a", false, "show all images")
	if err := cmd.Parse(args); err != nil {
		return nil
	}
	if cmd.NArg() > 1 {
		cmd.Usage()
		return nil
	}
	var nameFilter string
	if cmd.NArg() == 1 {
		nameFilter = cmd.Arg(0)
	}
	w := tabwriter.NewWriter(stdout, 20, 1, 3, ' ', 0)
	if !*quiet {
		fmt.Fprintf(w, "REPOSITORY\tTAG\tID\tCREATED\tPARENT\n")
	}
	var allImages map[string]*Image
	var err error
	if *fl_a {
		allImages, err = srv.runtime.graph.Map()
	} else {
		allImages, err = srv.runtime.graph.Heads()
	}
	if err != nil {
		return err
	}
	for name, repository := range srv.runtime.repositories.Repositories {
		if nameFilter != "" && name != nameFilter {
			continue
		}
		for tag, id := range repository {
			image, err := srv.runtime.graph.Get(id)
			if err != nil {
				log.Printf("Warning: couldn't load %s from %s/%s: %s", id, name, tag, err)
				continue
			}
			delete(allImages, id)
			if !*quiet {
				for idx, field := range []string{
					/* REPOSITORY */ name,
					/* TAG */ tag,
					/* ID */ id,
					/* CREATED */ HumanDuration(time.Now().Sub(image.Created)) + " ago",
					/* PARENT */ srv.runtime.repositories.ImageName(image.Parent),
				} {
					if idx == 0 {
						w.Write([]byte(field))
					} else {
						w.Write([]byte("\t" + field))
					}
				}
				w.Write([]byte{'\n'})
			} else {
				stdout.Write([]byte(image.Id + "\n"))
			}
		}
	}
	// Display images which aren't part of a
	if nameFilter == "" {
		for id, image := range allImages {
			if !*quiet {
				for idx, field := range []string{
					/* REPOSITORY */ "",
					/* TAG */ "",
					/* ID */ id,
					/* CREATED */ HumanDuration(time.Now().Sub(image.Created)) + " ago",
					/* PARENT */ srv.runtime.repositories.ImageName(image.Parent),
				} {
					if idx == 0 {
						w.Write([]byte(field))
					} else {
						w.Write([]byte("\t" + field))
					}
				}
				w.Write([]byte{'\n'})
			} else {
				stdout.Write([]byte(image.Id + "\n"))
			}
		}
	}
	if !*quiet {
		w.Flush()
	}
	return nil
}

В функции происходит простая итерация по полученным образам, фильтрация на основе имени репозитория и вывод полей в консольную таблицу. Map, Head и вспомогательные к ним функции, формирующие хеш таблицы allImages, находятся в файле graph.go:

graph.go

func (graph *Graph) Map() (map[string]*Image, error) {
	// FIXME: this should replace All()
	all, err := graph.All()
	if err != nil {
		return nil, err
	}
	images := make(map[string]*Image, len(all))
	for _, image := range all {
		images[image.Id] = image
	}
	return images, nil
}

func (graph *Graph) All() ([]*Image, error) {
	var images []*Image
	err := graph.WalkAll(func(image *Image) {
		images = append(images, image)
	})
	return images, err
}

func (graph *Graph) WalkAll(handler func(*Image)) error {
	files, err := ioutil.ReadDir(graph.Root)
	if err != nil {
		return err
	}
	for _, st := range files {
		if img, err := graph.Get(st.Name()); err != nil {
			// Skip image
			continue
		} else if handler != nil {
			handler(img)
		}
	}
	return nil
}

func (graph *Graph) ByParent() (map[string][]*Image, error) {
	byParent := make(map[string][]*Image)
	err := graph.WalkAll(func(image *Image) {
		image, err := graph.Get(image.Parent)
		if err != nil {
			return
		}
		if children, exists := byParent[image.Parent]; exists {
			byParent[image.Parent] = []*Image{image}
		} else {
			byParent[image.Parent] = append(children, image)
		}
	})
	return byParent, err
}

func (graph *Graph) Heads() (map[string]*Image, error) {
	heads := make(map[string]*Image)
	byParent, err := graph.ByParent()
	if err != nil {
		return nil, err
	}
	err = graph.WalkAll(func(image *Image) {
		// If it's not in the byParent lookup table, then
		// it's not a parent -> so it's a head!
		if _, exists := byParent[image.Id]; !exists {
			heads[image.Id] = image
		}
	})
	return heads, err
}

History

Отображает историю образа:

CmdHistory

func (srv *Server) CmdHistory(stdin io.ReadCloser, stdout io.Writer, args ...string) error {
	cmd := rcli.Subcmd(stdout, "history", "[OPTIONS] IMAGE", "Show the history of an image")
	if cmd.Parse(args) != nil || cmd.NArg() != 1 {
		cmd.Usage()
		return nil
	}
	image, err := srv.runtime.repositories.LookupImage(cmd.Arg(0))
	if err != nil {
		return err
	}
	w := tabwriter.NewWriter(stdout, 20, 1, 3, ' ', 0)
	defer w.Flush()
	fmt.Fprintf(w, "ID\tCREATED\tCREATED BY\n")
	return image.WalkHistory(func(img *Image) error {
		fmt.Fprintf(w, "%s\t%s\t%s\n",
			srv.runtime.repositories.ImageName(img.Id),
			HumanDuration(time.Now().Sub(img.Created))+" ago",
			strings.Join(img.ContainerConfig.Cmd, " "),
		)
		return nil
	})
}

func (img *Image) WalkHistory(handler func(*Image) error) (err error) {
    currentImg := img
    for currentImg != nil {
        if handler != nil {
            if err := handler(currentImg); err != nil {
                return err
            }
        }
        currentImg, err = currentImg.GetParent()
        if err != nil {
            return fmt.Errorf("Error while getting parent image: %v", err)
        }
    }
    return nil
}

func (img *Image) GetParent() (*Image, error) {
    if img.Parent == "" {
        return nil, nil
    }
    if img.graph == nil {
        return nil, fmt.Errorf("Can't lookup parent of unregistered image")
    }
    return img.graph.Get(img.Parent)
}

После получения структуры образа по переданному имени вызывается метод image.WalkHistory, который по цепочке обходит родительские образы, используя сохраненные ссылки Image.Parent, и выводит информацию в виде таблицы.

Commit

Создает новый образ на основе измененных данных файловой системы контейнера:

CmdCommit

func (srv *Server) CmdCommit(stdin io.ReadCloser, stdout io.Writer, args ...string) error {
	cmd := rcli.Subcmd(stdout,
		"commit", "[OPTIONS] CONTAINER [REPOSITORY [TAG]]",
		"Create a new image from a container's changes")
	if err := cmd.Parse(args); err != nil {
		return nil
	}
	containerName, repository, tag := cmd.Arg(0), cmd.Arg(1), cmd.Arg(2)
	if containerName == "" {
		cmd.Usage()
		return nil
	}
	img, err := srv.runtime.Commit(containerName, repository, tag)
	if err != nil {
		return err
	}
	fmt.Fprintln(stdout, img.Id)
	return nil
}

// Commit creates a new filesystem image from the current state of a container.
// The image can optionally be tagged into a repository
func (runtime *Runtime) Commit(id, repository, tag string) (*Image, error) {
	container := runtime.Get(id)
	if container == nil {
		return nil, fmt.Errorf("No such container: %s", id)
	}
	// FIXME: freeze the container before copying it to avoid data corruption?
	// FIXME: this shouldn't be in commands.
	rwTar, err := container.ExportRw()
	if err != nil {
		return nil, err
	}
	// Create a new image from the container's base layers + a new layer from container changes
	img, err := runtime.graph.Create(rwTar, container, "")
	if err != nil {
		return nil, err
	}
	// Register the image if needed
	if repository != "" {
		if err := runtime.repositories.Set(repository, tag, img.Id, true); err != nil {
			return img, err
		}
	}
	return img, nil
}

Метод Commit получает структуру container по переданному имени, вызывает метод container.ExportRw, который возвращает архив с директорией rw, после чего передает его в метод graph.Create, который мы уже разбирали выше. Если передано имя репозитория и tag, то дополнительно будет создан tag образа. Этот функционал будет разобран ниже в команде Tag.

Tag

Функция создает tag образа в локальном репозитории. Используется при импортировании образа и коммите контейнера:

CmdTag

func (srv *Server) CmdTag(stdin io.ReadCloser, stdout io.Writer, args ...string) error {
	cmd := rcli.Subcmd(stdout, "tag", "[OPTIONS] IMAGE REPOSITORY [TAG]", "Tag an image into a repository")
	force := cmd.Bool("f", false, "Force")
	if err := cmd.Parse(args); err != nil {
		return nil
	}
	if cmd.NArg() < 2 {
		cmd.Usage()
		return nil
	}
	return srv.runtime.repositories.Set(cmd.Arg(1), cmd.Arg(2), cmd.Arg(0), *force)
}

После обработки параметров вызывается метод repositories.Set из файла tags.go. Ниже я приведу содержание файла tags.go, который отвечает за весь этот функционал. Общий принцип работы довольно простой - структура TagStore имеет хеш таблицу(map) для соответствия тегов и образов. При добавлении нового тега он проходит валидацию на запрещенные символы и записывается в хеш таблицу. Далее структура TagStore экспортируется в json и сохраняется в файле на диске. При запуске докер структура загружается из этого файла и в дальнейшем на ее основе осуществляется поиск и фильтрация образов по именам и тегам:

tags.go

const DEFAULT_TAG = "latest"

type TagStore struct {
	path         string
	graph        *Graph
	Repositories map[string]Repository
}

type Repository map[string]string

func NewTagStore(path string, graph *Graph) (*TagStore, error) {
	abspath, err := filepath.Abs(path)
	if err != nil {
		return nil, err
	}
	store := &TagStore{
		path:         abspath,
		graph:        graph,
		Repositories: make(map[string]Repository),
	}
	// Load the json file if it exists, otherwise create it.
	if err := store.Reload(); os.IsNotExist(err) {
		if err := store.Save(); err != nil {
			return nil, err
		}
	} else if err != nil {
		return nil, err
	}
	return store, nil
}

func (store *TagStore) Save() error {
	// Store the json ball
	jsonData, err := json.Marshal(store)
	if err != nil {
		return err
	}
	if err := ioutil.WriteFile(store.path, jsonData, 0600); err != nil {
		return err
	}
	return nil
}

func (store *TagStore) Reload() error {
	jsonData, err := ioutil.ReadFile(store.path)
	if err != nil {
		return err
	}
	if err := json.Unmarshal(jsonData, store); err != nil {
		return err
	}
	return nil
}

func (store *TagStore) LookupImage(name string) (*Image, error) {
	img, err := store.graph.Get(name)
	if err != nil {
		// FIXME: standardize on returning nil when the image doesn't exist, and err for everything else
		// (so we can pass all errors here)
		repoAndTag := strings.SplitN(name, ":", 2)
		if len(repoAndTag) == 1 {
			repoAndTag = append(repoAndTag, DEFAULT_TAG)
		}
		if i, err := store.GetImage(repoAndTag[0], repoAndTag[1]); err != nil {
			return nil, err
		} else if i == nil {
			return nil, fmt.Errorf("No such image: %s", name)
		} else {
			img = i
		}
	}
	return img, nil
}

// Return a reverse-lookup table of all the names which refer to each image
// Eg. {"43b5f19b10584": {"base:latest", "base:v1"}}
func (store *TagStore) ById() map[string][]string {
	byId := make(map[string][]string)
	for repoName, repository := range store.Repositories {
		for tag, id := range repository {
			name := repoName + ":" + tag
			if _, exists := byId[id]; !exists {
				byId[id] = []string{name}
			} else {
				byId[id] = append(byId[id], name)
			}
		}
	}
	return byId
}

func (store *TagStore) ImageName(id string) string {
	if names, exists := store.ById()[id]; exists && len(names) > 0 {
		return names[0]
	}
	return id
}

func (store *TagStore) Set(repoName, tag, imageName string, force bool) error {
	img, err := store.LookupImage(imageName)
	if err != nil {
		return err
	}
	if tag == "" {
		tag = DEFAULT_TAG
	}
	if err := validateRepoName(repoName); err != nil {
		return err
	}
	if err := validateTagName(tag); err != nil {
		return err
	}
	if err := store.Reload(); err != nil {
		return err
	}
	var repo Repository
	if r, exists := store.Repositories[repoName]; exists {
		repo = r
	} else {
		repo = make(map[string]string)
		if old, exists := store.Repositories[repoName]; exists && !force {
			return fmt.Errorf("Tag %s:%s is already set to %s", repoName, tag, old)
		}
		store.Repositories[repoName] = repo
	}
	repo[tag] = img.Id
	return store.Save()
}

func (store *TagStore) Get(repoName string) (Repository, error) {
	if err := store.Reload(); err != nil {
		return nil, err
	}
	if r, exists := store.Repositories[repoName]; exists {
		return r, nil
	}
	return nil, nil
}

func (store *TagStore) GetImage(repoName, tag string) (*Image, error) {
	repo, err := store.Get(repoName)
	if err != nil {
		return nil, err
	} else if repo == nil {
		return nil, nil
	}
	if revision, exists := repo[tag]; exists {
		return store.graph.Get(revision)
	}
	return nil, nil
}

// Validate the name of a repository
func validateRepoName(name string) error {
	if name == "" {
		return fmt.Errorf("Repository name can't be empty")
	}
	if strings.Contains(name, ":") {
		return fmt.Errorf("Illegal repository name: %s", name)
	}
	return nil
}

// Validate the name of a tag
func validateTagName(name string) error {
	if name == "" {
		return fmt.Errorf("Tag name can't be empty")
	}
	if strings.Contains(name, "/") || strings.Contains(name, ":") {
		return fmt.Errorf("Illegal tag name: %s", name)
	}
	return nil
}

Управление контейнерами

Run

Создает и запускает контейнер на основе заданного образа:

CmdRun

func (srv *Server) CmdRun(stdin io.ReadCloser, stdout io.Writer, args ...string) error {
	config, err := ParseRun(args)
	if err != nil {
		return err
	}
	if config.Image == "" {
		return fmt.Errorf("Image not specified")
	}
	if len(config.Cmd) == 0 {
		return fmt.Errorf("Command not specified")
	}
	// Create new container
	container, err := srv.runtime.Create(config)
	if err != nil {
		return errors.New("Error creating container: " + err.Error())
	}
	if config.OpenStdin {
		cmd_stdin, err := container.StdinPipe()
		if err != nil {
			return err
		}
		if !config.Detach {
			Go(func() error {
				_, err := io.Copy(cmd_stdin, stdin)
				cmd_stdin.Close()
				return err
			})
		}
	}
	// Run the container
	if !config.Detach {
		cmd_stderr, err := container.StderrPipe()
		if err != nil {
			return err
		}
		cmd_stdout, err := container.StdoutPipe()
		if err != nil {
			return err
		}
		if err := container.Start(); err != nil {
			return err
		}
		sending_stdout := Go(func() error {
			_, err := io.Copy(stdout, cmd_stdout)
			return err
		})
		sending_stderr := Go(func() error {
			_, err := io.Copy(stdout, cmd_stderr)
			return err
		})
		err_sending_stdout := <-sending_stdout
		err_sending_stderr := <-sending_stderr
		if err_sending_stdout != nil {
			return err_sending_stdout
		}
		if err_sending_stderr != nil {
			return err_sending_stderr
		}
		container.Wait()
	} else {
		if err := container.Start(); err != nil {
			return err
		}
		fmt.Fprintln(stdout, container.Id)
	}
	return nil
}

В начале функция ParseRun производит разбор параметров и инициализацию структуры Config:

ParseRun

func ParseRun(args []string) (*Config, error) {
	cmd := flag.NewFlagSet("", flag.ContinueOnError)
	cmd.SetOutput(ioutil.Discard)
	fl_user := cmd.String("u", "", "Username or UID")
	fl_detach := cmd.Bool("d", false, "Detached mode: leave the container running in the background")
	fl_stdin := cmd.Bool("i", false, "Keep stdin open even if not attached")
	fl_tty := cmd.Bool("t", false, "Allocate a pseudo-tty")
	fl_memory := cmd.Int64("m", 0, "Memory limit (in bytes)")
	var fl_ports ports

	cmd.Var(&fl_ports, "p", "Map a network port to the container")
	var fl_env ListOpts
	cmd.Var(&fl_env, "e", "Set environment variables")
	if err := cmd.Parse(args); err != nil {
		return nil, err
	}
	config := &Config{
		Ports:     fl_ports,
		User:      *fl_user,
		Tty:       *fl_tty,
		OpenStdin: *fl_stdin,
		Memory:    *fl_memory,
		Detach:    *fl_detach,
		Env:       fl_env,
		Cmd:       cmd.Args()[1:],
		Image:     cmd.Arg(0),
	}
	return config, nil
}

На основании Config функция runtime.Create создает и возвращает новый контейнер, после чего при помощи пайпов и каналов перенаправляет потоки stdin, stdout, stderr, а затем запускает созданный контейнер методом container.Start.

runtime.Create

func (runtime *Runtime) Create(config *Config) (*Container, error) {
	// Lookup image
	img, err := runtime.repositories.LookupImage(config.Image)
	if err != nil {
		return nil, err
	}
	container := &Container{
		// FIXME: we should generate the ID here instead of receiving it as an argument
		Id:              GenerateId(),
		Created:         time.Now(),
		Path:            config.Cmd[0],
		Args:            config.Cmd[1:], //FIXME: de-duplicate from config
		Config:          config,
		Image:           img.Id, // Always use the resolved image id
		NetworkSettings: &NetworkSettings{},
		// FIXME: do we need to store this in the container?
		SysInitPath: sysInitPath,
	}
	container.root = runtime.containerRoot(container.Id)
	// Step 1: create the container directory.
	// This doubles as a barrier to avoid race conditions.
	if err := os.Mkdir(container.root, 0700); err != nil {
		return nil, err
	}
	// Step 2: save the container json
	if err := container.ToDisk(); err != nil {
		return nil, err
	}
	// Step 3: register the container
	if err := runtime.Register(container); err != nil {
		return nil, err
	}
	return container, nil
}

Здесь происходит инициализация структуры Container, создание рабочей директории контейнера и экспорт структуры в json формате. Завершает это вызов метода runtime.Register, код которого мы разбирали в прошлой статье. Функция GenerateId была рассмотрена ранее в разделе по команде import. Теперь перейдем к методу container.Start:

container.Start

func (container *Container) Start() error {
	if err := container.EnsureMounted(); err != nil {
		return err
	}
	if err := container.allocateNetwork(); err != nil {
		return err
	}
	if err := container.generateLXCConfig(); err != nil {
		return err
	}
	params := []string{
		"-n", container.Id,
		"-f", container.lxcConfigPath(),
		"--",
		"/sbin/init",
	}

	// Networking
	params = append(params, "-g", container.network.Gateway.String())

	// User
	if container.Config.User != "" {
		params = append(params, "-u", container.Config.User)
	}

	// Program
	params = append(params, "--", container.Path)
	params = append(params, container.Args...)

	container.cmd = exec.Command("/usr/bin/lxc-start", params...)

	// Setup environment
	container.cmd.Env = append(
		[]string{
			"HOME=/",
			"PATH=/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin",
		},
		container.Config.Env...,
	)

	var err error
	if container.Config.Tty {
		err = container.startPty()
	} else {
		err = container.start()
	}
	if err != nil {
		return err
	}
	// FIXME: save state on disk *first*, then converge
	// this way disk state is used as a journal, eg. we can restore after crash etc.
	container.State.setRunning(container.cmd.Process.Pid)
	container.ToDisk()
	go container.monitor()
	return nil
}

Основная логика его работы практически не изменилась со времен первого коммита, который мы разбирали в первой статье. Изменения коснулись части отвечающей за монтирование файловой системы, а также был добавлен код для сетевого стека и новый метод запуска процесса в контейнере. Разберем все по порядку, начиная с монтирования файловой системы:

EnsureMounted

func (container *Container) EnsureMounted() error {
	if mounted, err := container.Mounted(); err != nil {
		return err
	} else if mounted {
		return nil
	}
	return container.Mount()
} 

func (container *Container) Mounted() (bool, error) {
	return Mounted(container.RootfsPath())
} 

func Mounted(mountpoint string) (bool, error) {
	mntpoint, err := os.Stat(mountpoint)
	if err != nil {
		if os.IsNotExist(err) {
			return false, nil
		}
		return false, err
	}
	parent, err := os.Stat(filepath.Join(mountpoint, ".."))
	if err != nil {
		return false, err
	}
	mntpointSt := mntpoint.Sys().(*syscall.Stat_t)
	parentSt := parent.Sys().(*syscall.Stat_t)
	return mntpointSt.Dev != parentSt.Dev, nil
}

Метод container.EnsureMounted проверяет была ли смонтирована файловая система, в противном случае - выполняет монтирование вызовом container.Mount():

container.Mount

func (container *Container) Mount() error {
	image, err := container.GetImage()
	if err != nil {
		return err
	}
	return image.Mount(container.RootfsPath(), container.rwPath())
}

func (image *Image) Mount(root, rw string) error {
	if mounted, err := Mounted(root); err != nil {
		return err
	} else if mounted {
		return fmt.Errorf("%s is already mounted", root)
	}
	layers, err := image.layers()
	if err != nil {
		return err
	}
	// Create the target directories if they don't exist
	if err := os.Mkdir(root, 0755); err != nil && !os.IsExist(err) {
		return err
	}
	if err := os.Mkdir(rw, 0755); err != nil && !os.IsExist(err) {
		return err
	}
	// FIXME: @creack shouldn't we do this after going over changes?
	if err := MountAUFS(layers, rw, root); err != nil {
		return err
	}
	// FIXME: Create tests for deletion
	// FIXME: move this part to change.go
	// Retrieve the changeset from the parent and apply it to the container
	//  - Retrieve the changes
	changes, err := Changes(layers, layers[0])
	if err != nil {
		return err
	}
	// Iterate on changes
	for _, c := range changes {
		// If there is a delete
		if c.Kind == ChangeDelete {
			// Make sure the directory exists
			file_path, file_name := path.Dir(c.Path), path.Base(c.Path)
			if err := os.MkdirAll(path.Join(rw, file_path), 0755); err != nil {
				return err
			}
			// And create the whiteout (we just need to create empty file, discard the return)
			if _, err := os.Create(path.Join(path.Join(rw, file_path),
				".wh."+path.Base(file_name))); err != nil {
				return err
			}
		}
	}
	return nil
}

Подготовка параметров для монтирования производится функцией MountAUFS, логика которой аналогична первой версии, только теперь вместо утилиты mount монтирование производится системным вызовом:

MountAUFS

func MountAUFS(ro []string, rw string, target string) error {
	// FIXME: Now mount the layers
	rwBranch := fmt.Sprintf("%v=rw", rw)
	roBranches := ""
	for _, layer := range ro {
		roBranches += fmt.Sprintf("%v=ro:", layer)
	}
	branches := fmt.Sprintf("br:%v:%v", rwBranch, roBranches)
	return mount("none", target, "aufs", 0, branches)
}

func mount(source string, target string, fstype string, flags uintptr, data string) (err error) {
	return syscall.Mount(source, target, fstype, flags, data)
}

После монтирования функция получает изменения файловой системы с помощью команды Changes и производит создание файлов c .wh. префиксом в папке rw, если они были удалены в верхнем слое. Алгоритм работы этой функции будет рассмотрен в разделе команды Diff, вычисляющей изменения в файловой системе. Теперь перейдем к инициализации сети вызовом container.allocateNetwork:

container.allocateNetwork

func (container *Container) allocateNetwork() error {
	iface, err := container.runtime.networkManager.Allocate()
	if err != nil {
		return err
	}
	container.NetworkSettings.PortMapping = make(map[string]string)
	for _, port := range container.Config.Ports {
		if extPort, err := iface.AllocatePort(port); err != nil {
			iface.Release()
			return err
		} else {
			container.NetworkSettings.PortMapping[strconv.Itoa(port)] = strconv.Itoa(extPort)
		}
	}
	container.network = iface
	container.NetworkSettings.IpAddress = iface.IPNet.IP.String()
	container.NetworkSettings.IpPrefixLen, _ = iface.IPNet.Mask.Size()
	container.NetworkSettings.Gateway = iface.Gateway.String()
	return nil
}

В нем происходит настройка сетевого интерфейса, присвоение ip адреса, маски и шлюза, а также проброс портов. Я подробно разберу весь этот функционал ниже, в отдельной части по работе с сетевым стеком. Далее идет вызов метода generateLXCConfig. Он был подробно разобран в первой статье и остался практически без изменений. Стоит лишь отметить, что теперь в lxc_template.go добавлены настройки сети, монтирование /etc/resolv.conf для работы dns и, главное, монтирование исполняемого файла docker в точку /sbin/init, так как теперь выполнение процесса будет начинаться с него. Я уже обращал на это внимание в части 2.1. Ниже приведены изменения в lxc_template:

lxc_template.go

# network configuration
lxc.network.type = veth
lxc.network.flags = up
lxc.network.link = lxcbr0
lxc.network.name = eth0
lxc.network.mtu = 1500
lxc.network.ipv4 = {{.NetworkSettings.IpAddress}}/{{.NetworkSettings.IpPrefixLen}}

# Inject docker-init
lxc.mount.entry = {{.SysInitPath}} {{$ROOTFS}}/sbin/init none bind,ro 0 0

# In order to get a working DNS environment, mount bind (ro) the host's /etc/resolv.conf into the container
lxc.mount.entry = /etc/resolv.conf {{$ROOTFS}}/etc/resolv.conf none bind,ro 0 0

Теперь осталось разобрать файл sysinit.go, с которого стартует созданный lxc контейнер:

sysinit.go

// Setup networking
func setupNetworking(gw string) {
	if gw == "" {
		return
	}
	cmd := exec.Command("/sbin/route", "add", "default", "gw", gw)
	if err := cmd.Run(); err != nil {
		log.Fatalf("Unable to set up networking: %v", err)
	}
}

// Takes care of dropping privileges to the desired user
func changeUser(u string) {
	if u == "" {
		return
	}
	userent, err := user.LookupId(u)
	if err != nil {
		userent, err = user.Lookup(u)
	}
	if err != nil {
		log.Fatalf("Unable to find user %v: %v", u, err)
	}

	uid, err := strconv.Atoi(userent.Uid)
	if err != nil {
		log.Fatalf("Invalid uid: %v", userent.Uid)
	}
	gid, err := strconv.Atoi(userent.Gid)
	if err != nil {
		log.Fatalf("Invalid gid: %v", userent.Gid)
	}

	if err := syscall.Setgid(gid); err != nil {
		log.Fatalf("setgid failed: %v", err)
	}
	if err := syscall.Setuid(uid); err != nil {
		log.Fatalf("setuid failed: %v", err)
	}
}

func executeProgram(name string, args []string) {
	path, err := exec.LookPath(name)
	if err != nil {
		log.Printf("Unable to locate %v", name)
		os.Exit(127)
	}

	if err := syscall.Exec(path, args, os.Environ()); err != nil {
		panic(err)
	}
}

// Sys Init code
// This code is run INSIDE the container and is responsible for setting
// up the environment before running the actual process
func SysInit() {
	if len(os.Args) <= 1 {
		fmt.Println("You should not invoke docker-init manually")
		os.Exit(1)
	}
	var u = flag.String("u", "", "username or uid")
	var gw = flag.String("g", "", "gateway address")

	flag.Parse()

	setupNetworking(*gw)
	changeUser(*u)
	executeProgram(flag.Arg(0), flag.Args())
}

Как видим, в SysInit происходит настройка окружения перед запуском процесса. Добавление default gateway в таблицу маршрутизации, настройка пользователя и группы, под которыми будет выполняться процесс, и собственно запуск процесса стандартным методом Exec.

После запуска процесса в методе Start идет перенаправление стандартных потоков и запуск горутины container.monitor, работу которой мы разбирали в первой статье. Можно лишь добавить, что теперь в ней происходит освобождение назначенного ip адреса, проброшенных портов и размонтирование файловой системы.

container.monitor

func (container *Container) monitor() {
	// Wait for the program to exit
	container.cmd.Wait()
	exitCode := container.cmd.ProcessState.Sys().(syscall.WaitStatus).ExitStatus()

	// Cleanup
	if err := container.releaseNetwork(); err != nil {
		log.Printf("%v: Failed to release network: %v", container.Id, err)
	}
	container.stdout.Close()
	container.stderr.Close()
	if err := container.Unmount(); err != nil {
		log.Printf("%v: Failed to umount filesystem: %v", container.Id, err)
	}

	// Re-create a brand new stdin pipe once the container exited
	if container.Config.OpenStdin {
		container.stdin, container.stdinPipe = io.Pipe()
	}

	// Report status back
	container.State.setStopped(exitCode)
	container.ToDisk()
}

func (container *Container) releaseNetwork() error {
	err := container.network.Release()
	container.network = nil
	container.NetworkSettings = &NetworkSettings{}
	return err
}

func (container *Container) Unmount() error {
	return Unmount(container.RootfsPath())
}

func Unmount(target string) error {
	if err := syscall.Unmount(target, 0); err != nil {
		return err
	}
	// Even though we just unmounted the filesystem, AUFS will prevent deleting the mntpoint
	// for some time. We'll just keep retrying until it succeeds.
	for retries := 0; retries < 1000; retries++ {
		err := os.Remove(target)
		if err == nil {
			// rm mntpoint succeeded
			return nil
		}
		if os.IsNotExist(err) {
			// mntpoint doesn't exist anymore. Success.
			return nil
		}
		// fmt.Printf("(%v) Remove %v returned: %v\n", retries, target, err)
		time.Sleep(10 * time.Millisecond)
	}
	return fmt.Errorf("Umount: Failed to umount %v", target)
}

Неожиданное завершение

Ввиду того, что редактор хабра начал сильно тормозить, а местами и полностью зависать (видимо сказывается большое количество снипетов кода), я вынужден завершить эту часть, но продолжу в следующей, прямо с этого места.

Эволюция Docker. Часть 2.2