func runClusterMigrateDomain(args *docopt.Args) error {
	client, err := getClusterClient()
	if err != nil {
		shutdown.Fatal(err)
	}

	dm := &ct.DomainMigration{
		Domain: args.String["<domain>"],
	}

	release, err := client.GetAppRelease("controller")
	if err != nil {
		return err
	}
	dm.OldDomain = release.Env["DEFAULT_ROUTE_DOMAIN"]

	if !promptYesNo(fmt.Sprintf("Migrate cluster domain from %q to %q?", dm.OldDomain, dm.Domain)) {
		fmt.Println("Aborted")
		return nil
	}

	maxDuration := 2 * time.Minute
	fmt.Printf("Migrating cluster domain (this can take up to %s)...\n", maxDuration)

	events := make(chan *ct.Event)
	stream, err := client.StreamEvents(controller.StreamEventsOptions{
		ObjectTypes: []ct.EventType{ct.EventTypeDomainMigration},
	}, events)
	if err != nil {
		return nil
	}
	defer stream.Close()

	if err := client.PutDomain(dm); err != nil {
		return err
	}

	timeout := time.After(maxDuration)
	for {
		select {
		case event, ok := <-events:
			if !ok {
				return stream.Err()
			}
			var e *ct.DomainMigrationEvent
			if err := json.Unmarshal(event.Data, &e); err != nil {
				return err
			}
			if e.Error != "" {
				fmt.Println(e.Error)
			}
			if e.DomainMigration.FinishedAt != nil {
				fmt.Printf("Changed cluster domain from %q to %q\n", dm.OldDomain, dm.Domain)
				return nil
			}
		case <-timeout:
			return errors.New("timed out waiting for domain migration to complete")
		}
	}
}

func (f *ClusterFixer) FixController(instances []*discoverd.Instance, startScheduler bool) error {
	f.l.Info("found controller instance, checking critical formations")
	inst := instances[0]
	client, err := controller.NewClient("http://"+inst.Addr, inst.Meta["AUTH_KEY"])
	if err != nil {
		return fmt.Errorf("unexpected error creating controller client: %s", err)
	}

	// check that formations for critical components are expected
	apps := []string{"controller", "router", "discoverd", "flannel", "postgres"}
	changes := make(map[string]*ct.Formation, len(apps))
	var controllerFormation *ct.Formation
	for _, app := range apps {
		release, err := client.GetAppRelease(app)
		if err != nil {
			return fmt.Errorf("error getting %s release: %s", app, err)
		}
		formation, err := client.GetFormation(app, release.ID)
		if err != nil {
			// TODO: handle ErrNotFound
			return fmt.Errorf("error getting %s formation: %s", app, err)
		}
		if app == "controller" {
			controllerFormation = formation
		}
		for typ := range release.Processes {
			var want int
			if app == "postgres" && typ == "postgres" && len(f.hosts) > 1 && formation.Processes[typ] < 3 {
				want = 3
			} else if formation.Processes[typ] < 1 {
				want = 1
			}
			if want > 0 {
				f.l.Info("found broken formation", "app", app, "process", typ)
				if _, ok := changes[app]; !ok {
					if formation.Processes == nil {
						formation.Processes = make(map[string]int)
					}
					changes[app] = formation
				}
				changes[app].Processes[typ] = want
			}
		}
	}

	for app, formation := range changes {
		f.l.Info("fixing broken formation", "app", app)
		if err := client.PutFormation(formation); err != nil {
			return fmt.Errorf("error putting %s formation: %s", app, err)
		}
	}

	if startScheduler {
		if err := f.StartScheduler(client, controllerFormation); err != nil {
			return err
		}
	}
	return nil
}

func (s *CLISuite) TestDockerPush(t *c.C) {
	// build image with ENV and CMD
	repo := "cli-test-push"
	s.buildDockerImage(t, repo,
		`ENV FOO=BAR`,
		`CMD ["/bin/pingserv"]`,
	)

	// create app
	client := s.controllerClient(t)
	app := &ct.App{Name: "cli-test-docker-push"}
	t.Assert(client.CreateApp(app), c.IsNil)

	// flynn docker push image
	t.Assert(flynn(t, "/", "-a", app.Name, "docker", "push", repo), Succeeds)

	// check app was released with correct env, meta and process type
	release, err := client.GetAppRelease(app.ID)
	t.Assert(err, c.IsNil)
	t.Assert(release.Env["FOO"], c.Equals, "BAR")
	t.Assert(release.Meta["docker-receive"], c.Equals, "true")
	t.Assert(release.Processes, c.HasLen, 1)
	proc, ok := release.Processes["app"]
	if !ok {
		t.Fatal(`release missing "app" process type`)
	}
	t.Assert(proc.Args, c.DeepEquals, []string{"/bin/pingserv"})

	// check updated env vars are not overwritten
	//
	// need to remove the tag before pushing as we are using Docker 1.9
	// which does not overwrite tags.
	// TODO: remove this when upgrading Docker > 1.9
	u, err := url.Parse(s.clusterConf(t).DockerPushURL)
	t.Assert(err, c.IsNil)
	tag := fmt.Sprintf("%s/%s:latest", u.Host, app.Name)
	t.Assert(run(t, exec.Command("docker", "rmi", tag)), Succeeds)
	t.Assert(flynn(t, "/", "-a", app.Name, "env", "set", "FOO=BAZ"), Succeeds)
	t.Assert(flynn(t, "/", "-a", app.Name, "docker", "push", repo), Succeeds)
	t.Assert(flynn(t, "/", "-a", app.Name, "env", "get", "FOO"), Outputs, "BAZ\n")

	// check the release can be scaled up
	t.Assert(flynn(t, "/", "-a", app.Name, "scale", "app=1"), Succeeds)

	// check the job is reachable with the app's name in discoverd
	instances, err := s.discoverdClient(t).Instances(app.Name+"-web", 10*time.Second)
	t.Assert(err, c.IsNil)
	res, err := hh.RetryClient.Get("http://" + instances[0].Addr)
	t.Assert(err, c.IsNil)
	defer res.Body.Close()
	body, err := ioutil.ReadAll(res.Body)
	t.Assert(err, c.IsNil)
	t.Assert(string(body), c.Equals, "OK")
}

// CheckScale examines sirenia cluster formation to check if cluster
// has been scaled up yet.
// Returns true if scaled, false if not.
func CheckScale(app, controllerKey, procName string, logger log15.Logger) (bool, error) {
	logger = logger.New("fn", "CheckScale")
	// Connect to controller.
	logger.Info("connecting to controller")
	client, err := controller.NewClient("", controllerKey)
	if err != nil {
		logger.Error("controller client error", "err", err)
		return false, err
	}

	// Retrieve app release.
	logger.Info("retrieving app release", "app", app)
	release, err := client.GetAppRelease(app)
	if err == controller.ErrNotFound {
		logger.Error("release not found", "app", app)
		return false, err
	} else if err != nil {
		logger.Error("get release error", "app", app, "err", err)
		return false, err
	}

	// Retrieve current formation.
	logger.Info("retrieving formation", "app", app, "release_id", release.ID)
	formation, err := client.GetFormation(app, release.ID)
	if err == controller.ErrNotFound {
		logger.Error("formation not found", "app", app, "release_id", release.ID)
		return false, err
	} else if err != nil {
		logger.Error("formation error", "app", app, "release_id", release.ID, "err", err)
		return false, err
	}

	// Database hasn't been scaled up yet
	if formation.Processes[procName] == 0 {
		return false, nil
	}

	return true, nil

}

func (s *CLISuite) TestDockerExportImport(t *c.C) {
	// release via docker-receive
	client := s.controllerClient(t)
	app := &ct.App{Name: "cli-test-docker-export"}
	t.Assert(client.CreateApp(app), c.IsNil)
	repo := "cli-test-export"
	s.buildDockerImage(t, repo, `CMD ["/bin/pingserv"]`)
	t.Assert(flynn(t, "/", "-a", app.Name, "docker", "push", repo), Succeeds)
	t.Assert(flynn(t, "/", "-a", app.Name, "scale", "app=1"), Succeeds)
	defer flynn(t, "/", "-a", app.Name, "scale", "app=0")

	// export the app
	file := filepath.Join(t.MkDir(), "export.tar")
	t.Assert(flynn(t, "/", "-a", app.Name, "export", "-f", file), Succeeds)

	// delete the image from the registry
	release, err := client.GetAppRelease(app.Name)
	t.Assert(err, c.IsNil)
	artifact, err := client.GetArtifact(release.ImageArtifactID())
	t.Assert(err, c.IsNil)
	u, err := url.Parse(s.clusterConf(t).DockerPushURL)
	t.Assert(err, c.IsNil)
	uri := fmt.Sprintf("http://%s/v2/%s/manifests/%s", u.Host, app.Name, artifact.Meta["docker-receive.digest"])
	req, err := http.NewRequest("DELETE", uri, nil)
	req.SetBasicAuth("", s.clusterConf(t).Key)
	t.Assert(err, c.IsNil)
	res, err := http.DefaultClient.Do(req)
	t.Assert(err, c.IsNil)
	res.Body.Close()

	// import to another app
	importApp := "cli-test-docker-import"
	t.Assert(flynn(t, "/", "import", "--name", importApp, "--file", file), Succeeds)
	defer flynn(t, "/", "-a", importApp, "scale", "app=0")

	// wait for it to start
	_, err = s.discoverdClient(t).Instances(importApp+"-web", 10*time.Second)
	t.Assert(err, c.IsNil)
}

func runClusterMigrateDomain(args *docopt.Args) error {
	cluster, err := getCluster()
	if err != nil {
		shutdown.Fatal(err)
	}
	client, err := cluster.Client()
	if err != nil {
		shutdown.Fatal(err)
	}

	dm := &ct.DomainMigration{
		Domain: args.String["<domain>"],
	}

	release, err := client.GetAppRelease("controller")
	if err != nil {
		return err
	}
	dm.OldDomain = release.Env["DEFAULT_ROUTE_DOMAIN"]

	if !promptYesNo(fmt.Sprintf("Migrate cluster domain from %q to %q?", dm.OldDomain, dm.Domain)) {
		fmt.Println("Aborted")
		return nil
	}

	maxDuration := 2 * time.Minute
	fmt.Printf("Migrating cluster domain (this can take up to %s)...\n", maxDuration)

	events := make(chan *ct.Event)
	stream, err := client.StreamEvents(ct.StreamEventsOptions{
		ObjectTypes: []ct.EventType{ct.EventTypeDomainMigration},
	}, events)
	if err != nil {
		return nil
	}
	defer stream.Close()

	if err := client.PutDomain(dm); err != nil {
		return err
	}

	timeout := time.After(maxDuration)
	for {
		select {
		case event, ok := <-events:
			if !ok {
				return stream.Err()
			}
			var e *ct.DomainMigrationEvent
			if err := json.Unmarshal(event.Data, &e); err != nil {
				return err
			}
			if e.Error != "" {
				fmt.Println(e.Error)
			}
			if e.DomainMigration.FinishedAt != nil {
				dm = e.DomainMigration
				fmt.Printf("Changed cluster domain from %q to %q\n", dm.OldDomain, dm.Domain)

				// update flynnrc
				cluster.TLSPin = dm.TLSCert.Pin
				cluster.ControllerURL = fmt.Sprintf("https://controller.%s", dm.Domain)
				cluster.GitURL = fmt.Sprintf("https://git.%s", dm.Domain)
				cluster.DockerPushURL = fmt.Sprintf("https://docker.%s", dm.Domain)
				if err := config.SaveTo(configPath()); err != nil {
					return fmt.Errorf("Error saving config: %s", err)
				}

				// update git config
				caFile, err := cfg.CACertFile(cluster.Name)
				if err != nil {
					return err
				}
				defer caFile.Close()
				if _, err := caFile.Write([]byte(dm.TLSCert.CACert)); err != nil {
					return err
				}
				if err := cfg.WriteGlobalGitConfig(cluster.GitURL, caFile.Name()); err != nil {
					return err
				}
				cfg.RemoveGlobalGitConfig(fmt.Sprintf("https://git.%s", dm.OldDomain))

				// try to run "docker login" for the new domain, but just print a warning
				// if it fails so the user can fix it later
				if host, err := cluster.DockerPushHost(); err == nil {
					if err := dockerLogin(host, cluster.Key); err == ErrDockerTLSError {
						printDockerTLSWarning(host, caFile.Name())
					}
				}
				dockerLogout(dm.OldDomain)

				fmt.Println("Updated local CLI configuration")
				return nil
			}
		case <-timeout:
			return errors.New("timed out waiting for domain migration to complete")
		}
	}
}

// ScaleUp scales up a dormant Sirenia cluster
func ScaleUp(app, controllerKey, serviceAddr, procName, singleton string, logger log15.Logger) error {
	logger = logger.New("fn", "ScaleUp")
	sc := sirenia.NewClient(serviceAddr)

	logger.Info("checking status", "host", serviceAddr)
	if status, err := sc.Status(); err == nil && status.Database != nil && status.Database.ReadWrite {
		logger.Info("database is up, skipping scale")
		// Skip the rest, the database is already available
		return nil
	} else if err != nil {
		logger.Info("error checking status", "err", err)
	} else {
		logger.Info("got status, but database is not read-write")
	}

	// Connect to controller.
	logger.Info("connecting to controller")
	client, err := controller.NewClient("", controllerKey)
	if err != nil {
		logger.Error("controller client error", "err", err)
		return err
	}

	// Retrieve the app release.
	logger.Info("retrieving app release", "app", app)
	release, err := client.GetAppRelease(app)
	if err == controller.ErrNotFound {
		logger.Error("release not found", "app", app)
		return errors.New("release not found")
	} else if err != nil {
		logger.Error("get release error", "app", app, "err", err)
		return err
	}

	// Retrieve current formation.
	logger.Info("retrieving formation", "app", app, "release_id", release.ID)
	formation, err := client.GetFormation(app, release.ID)
	if err == controller.ErrNotFound {
		logger.Error("formation not found", "app", app, "release_id", release.ID)
		return errors.New("formation not found")
	} else if err != nil {
		logger.Error("formation error", "app", app, "release_id", release.ID, "err", err)
		return err
	}

	// If database is running then exit.
	if formation.Processes[procName] > 0 {
		logger.Info("database is running, scaling not necessary")
		return nil
	}

	// Copy processes and increase database processes.
	processes := make(map[string]int, len(formation.Processes))
	for k, v := range formation.Processes {
		processes[k] = v
	}

	if singleton == "true" {
		processes[procName] = 1
	} else {
		processes[procName] = 3
	}

	// Update formation.
	logger.Info("updating formation", "app", app, "release_id", release.ID)
	formation.Processes = processes
	if err := client.PutFormation(formation); err != nil {
		logger.Error("put formation error", "app", app, "release_id", release.ID, "err", err)
		return err
	}

	if err := sc.WaitForReadWrite(5 * time.Minute); err != nil {
		logger.Error("wait for read write", "err", err)
		return errors.New("timed out while starting sirenia cluster")
	}

	logger.Info("scaling complete")
	return nil
}

func runClusterBackup(args *docopt.Args) error {
	client, err := getClusterClient()
	if err != nil {
		return err
	}

	var bar *pb.ProgressBar
	if term.IsTerminal(os.Stderr.Fd()) {
		bar = pb.New(0)
		bar.SetUnits(pb.U_BYTES)
		bar.ShowBar = false
		bar.ShowSpeed = true
		bar.Output = os.Stderr
		bar.Start()
	}

	var dest io.Writer = os.Stdout
	if filename := args.String["--file"]; filename != "" {
		f, err := os.Create(filename)
		if err != nil {
			return err
		}
		defer f.Close()
		dest = f
	}

	fmt.Fprintln(os.Stderr, "Creating cluster backup...")

	tw := NewTarWriter("flynn-backup-"+time.Now().UTC().Format("2006-01-02_150405"), dest)
	defer tw.Close()

	// get app and release details for key apps
	data := make(map[string]*ct.ExpandedFormation, 4)
	for _, name := range []string{"postgres", "discoverd", "flannel", "controller"} {
		app, err := client.GetApp(name)
		if err != nil {
			return fmt.Errorf("error getting %s app details: %s", name, err)
		}
		release, err := client.GetAppRelease(app.ID)
		if err != nil {
			return fmt.Errorf("error getting %s app release: %s", name, err)
		}
		formation, err := client.GetFormation(app.ID, release.ID)
		if err != nil {
			return fmt.Errorf("error getting %s app formation: %s", name, err)
		}
		artifact, err := client.GetArtifact(release.ArtifactID)
		if err != nil {
			return fmt.Errorf("error getting %s app artifact: %s", name, err)
		}
		data[name] = &ct.ExpandedFormation{
			App:       app,
			Release:   release,
			Artifact:  artifact,
			Processes: formation.Processes,
		}
	}
	if err := tw.WriteJSON("flynn.json", data); err != nil {
		return err
	}

	config := &runConfig{
		App:        "postgres",
		Release:    data["postgres"].Release.ID,
		Entrypoint: []string{"sh"},
		Args:       []string{"-c", "pg_dumpall --clean --if-exists | gzip -9"},
		Env: map[string]string{
			"PGHOST":     "leader.postgres.discoverd",
			"PGUSER":     "flynn",
			"PGPASSWORD": data["postgres"].Release.Env["PGPASSWORD"],
		},
		DisableLog: true,
	}
	if err := tw.WriteCommandOutput(client, "postgres.sql.gz", config, bar); err != nil {
		return fmt.Errorf("error dumping database: %s", err)
	}

	if bar != nil {
		bar.Finish()
	}
	fmt.Fprintln(os.Stderr, "Backup complete.")

	return nil
}

func (s *DomainMigrationSuite) migrateDomain(t *c.C, dm *ct.DomainMigration) {
	debugf(t, "migrating domain from %s to %s", dm.OldDomain, dm.Domain)
	client := s.controllerClient(t)

	events := make(chan *ct.Event)
	stream, err := client.StreamEvents(controller.StreamEventsOptions{
		ObjectTypes: []ct.EventType{ct.EventTypeDomainMigration},
	}, events)
	t.Assert(err, c.IsNil)
	defer stream.Close()

	prevRouterRelease, err := client.GetAppRelease("router")
	t.Assert(err, c.IsNil)

	err = client.PutDomain(dm)
	t.Assert(err, c.IsNil)

	waitEvent := func(typ string, timeout time.Duration) (event ct.DomainMigrationEvent) {
		debugf(t, "waiting for %s domain migration event", typ)
		var e *ct.Event
		var ok bool
		select {
		case e, ok = <-events:
			if !ok {
				t.Fatal("event stream closed unexpectedly")
			}
			debugf(t, "got %s domain migration event", typ)
		case <-time.After(timeout):
			t.Fatalf("timed out waiting for %s domain migration event", typ)
		}
		t.Assert(e.Data, c.NotNil)
		t.Assert(json.Unmarshal(e.Data, &event), c.IsNil)
		return
	}

	// created
	event := waitEvent("initial", 2*time.Minute)
	t.Assert(event.Error, c.Equals, "")
	t.Assert(event.DomainMigration, c.NotNil)
	t.Assert(event.DomainMigration.ID, c.Equals, dm.ID)
	t.Assert(event.DomainMigration.OldDomain, c.Equals, dm.OldDomain)
	t.Assert(event.DomainMigration.Domain, c.Equals, dm.Domain)
	t.Assert(event.DomainMigration.TLSCert, c.IsNil)
	t.Assert(event.DomainMigration.OldTLSCert, c.NotNil)
	t.Assert(event.DomainMigration.CreatedAt, c.NotNil)
	t.Assert(event.DomainMigration.CreatedAt.Equal(*dm.CreatedAt), c.Equals, true)
	t.Assert(event.DomainMigration.FinishedAt, c.IsNil)

	// complete
	event = waitEvent("final", 3*time.Minute)
	t.Assert(event.Error, c.Equals, "")
	t.Assert(event.DomainMigration, c.NotNil)
	t.Assert(event.DomainMigration.ID, c.Equals, dm.ID)
	t.Assert(event.DomainMigration.OldDomain, c.Equals, dm.OldDomain)
	t.Assert(event.DomainMigration.Domain, c.Equals, dm.Domain)
	t.Assert(event.DomainMigration.TLSCert, c.NotNil)
	t.Assert(event.DomainMigration.OldTLSCert, c.NotNil)
	t.Assert(event.DomainMigration.CreatedAt, c.NotNil)
	t.Assert(event.DomainMigration.CreatedAt.Equal(*dm.CreatedAt), c.Equals, true)
	t.Assert(event.DomainMigration.FinishedAt, c.NotNil)

	cert := event.DomainMigration.TLSCert

	controllerRelease, err := client.GetAppRelease("controller")
	t.Assert(err, c.IsNil)
	t.Assert(controllerRelease.Env["DEFAULT_ROUTE_DOMAIN"], c.Equals, dm.Domain)
	t.Assert(controllerRelease.Env["CA_CERT"], c.Equals, cert.CACert)

	routerRelease, err := client.GetAppRelease("router")
	t.Assert(err, c.IsNil)
	t.Assert(routerRelease.Env["TLSCERT"], c.Equals, cert.Cert)
	t.Assert(routerRelease.Env["TLSKEY"], c.Not(c.Equals), "")
	t.Assert(routerRelease.Env["TLSKEY"], c.Not(c.Equals), prevRouterRelease.Env["TLSKEY"])

	dashboardRelease, err := client.GetAppRelease("dashboard")
	t.Assert(err, c.IsNil)
	t.Assert(dashboardRelease.Env["DEFAULT_ROUTE_DOMAIN"], c.Equals, dm.Domain)
	t.Assert(dashboardRelease.Env["CONTROLLER_DOMAIN"], c.Equals, fmt.Sprintf("controller.%s", dm.Domain))
	t.Assert(dashboardRelease.Env["URL"], c.Equals, fmt.Sprintf("dashboard.%s", dm.Domain))
	t.Assert(dashboardRelease.Env["CA_CERT"], c.Equals, cert.CACert)

	var doPing func(string, int)
	doPing = func(component string, retriesRemaining int) {
		url := fmt.Sprintf("http://%s.%s/ping", component, dm.Domain)
		res, err := (&http.Client{}).Get(url)
		if (err != nil || res.StatusCode != 200) && retriesRemaining > 0 {
			time.Sleep(100 * time.Millisecond)
			doPing(component, retriesRemaining-1)
			return
		}
		t.Assert(err, c.IsNil)
		t.Assert(res.StatusCode, c.Equals, 200, c.Commentf("failed to ping %s", component))
	}
	doPing("controller", 3)
	doPing("dashboard", 3)
}

//.........这里部分代码省略.........
	ch <- &bootstrap.StepInfo{StepMeta: meta, State: "done", Timestamp: time.Now().UTC()}

	var mysqldb io.Reader
	if data.MariaDB != nil {
		mysqldb, err = getFile("mysql.sql.gz")
		if err != nil {
			return err
		}
	}

	// load data into mariadb if it was present in the backup.
	if mysqldb != nil && data.MariaDB != nil {
		cmd = exec.JobUsingHost(state.Hosts[0], host.Artifact{Type: data.MariaDB.Artifact.Type, URI: data.MariaDB.Artifact.URI}, nil)
		cmd.Entrypoint = []string{"mysql"}
		cmd.Cmd = []string{"-u", "flynn", "-h", "leader.mariadb.discoverd"}
		cmd.Env = map[string]string{
			"MYSQL_PWD": data.MariaDB.Release.Env["MYSQL_PWD"],
		}
		cmd.Stdin = mysqldb
		meta = bootstrap.StepMeta{ID: "restore", Action: "restore-mariadb"}
		ch <- &bootstrap.StepInfo{StepMeta: meta, State: "start", Timestamp: time.Now().UTC()}
		out, err = cmd.CombinedOutput()
		if os.Getenv("DEBUG") != "" {
			fmt.Println(string(out))
		}
		if err != nil {
			ch <- &bootstrap.StepInfo{
				StepMeta:  meta,
				State:     "error",
				Error:     fmt.Sprintf("error running mysql restore: %s - %q", err, string(out)),
				Err:       err,
				Timestamp: time.Now().UTC(),
			}
			return err
		}
		ch <- &bootstrap.StepInfo{StepMeta: meta, State: "done", Timestamp: time.Now().UTC()}
	}

	// start controller API
	data.Controller.Processes = map[string]int{"web": 1}
	_, err = bootstrap.Manifest{
		step("controller", "run-app", &bootstrap.RunAppAction{
			ExpandedFormation: data.Controller,
		}),
	}.RunWithState(ch, state)

	// wait for controller to come up
	meta = bootstrap.StepMeta{ID: "wait-controller", Action: "wait-controller"}
	ch <- &bootstrap.StepInfo{StepMeta: meta, State: "start", Timestamp: time.Now().UTC()}
	controllerInstances, err := discoverd.GetInstances("controller", 30*time.Second)
	if err != nil {
		return fmt.Errorf("error getting controller instance: %s", err)
	}

	// get blobstore config
	client, err := controller.NewClient("http://"+controllerInstances[0].Addr, data.Controller.Release.Env["AUTH_KEY"])
	if err != nil {
		return err
	}
	blobstoreRelease, err := client.GetAppRelease("blobstore")
	if err != nil {
		return fmt.Errorf("error getting blobstore release: %s", err)
	}
	blobstoreFormation, err := client.GetExpandedFormation("blobstore", blobstoreRelease.ID)
	if err != nil {
		return fmt.Errorf("error getting blobstore expanded formation: %s", err)
	}
	state.SetControllerKey(data.Controller.Release.Env["AUTH_KEY"])
	ch <- &bootstrap.StepInfo{StepMeta: meta, State: "done", Timestamp: time.Now().UTC()}

	// start blobstore, scheduler, and enable cluster monitor
	data.Controller.Processes = map[string]int{"scheduler": 1}
	// only start one scheduler instance
	schedulerProcess := data.Controller.Release.Processes["scheduler"]
	schedulerProcess.Omni = false
	data.Controller.Release.Processes["scheduler"] = schedulerProcess
	_, err = bootstrap.Manifest{
		step("blobstore", "run-app", &bootstrap.RunAppAction{
			ExpandedFormation: blobstoreFormation,
		}),
		step("blobstore-wait", "wait", &bootstrap.WaitAction{
			URL:    "http://blobstore.discoverd",
			Status: 404,
		}),
		step("controller-scheduler", "run-app", &bootstrap.RunAppAction{
			ExpandedFormation: data.Controller,
		}),
		step("status", "status-check", &bootstrap.StatusCheckAction{
			URL: "http://status-web.discoverd",
		}),
		step("cluster-monitor", "cluster-monitor", &bootstrap.ClusterMonitorAction{
			Enabled: true,
		}),
	}.RunWithState(ch, state)
	if err != nil {
		return err
	}

	return nil
}

func run() error {
	client, err := controller.NewClient("", os.Getenv("CONTROLLER_KEY"))
	if err != nil {
		return fmt.Errorf("Unable to connect to controller: %s", err)
	}

	usage := `
Usage: flynn-receiver <app> <rev> [-e <var>=<val>]... [-m <key>=<val>]...

Options:
	-e,--env <var>=<val>
	-m,--meta <key>=<val>
`[1:]
	args, _ := docopt.Parse(usage, nil, true, version.String(), false)

	appName := args.String["<app>"]
	env, err := parsePairs(args, "--env")
	if err != nil {
		return err
	}
	meta, err := parsePairs(args, "--meta")
	if err != nil {
		return err
	}

	slugBuilder, err := client.GetArtifact(os.Getenv("SLUGBUILDER_IMAGE_ID"))
	if err != nil {
		return fmt.Errorf("Error getting slugbuilder image: %s", err)
	}

	slugRunnerID := os.Getenv("SLUGRUNNER_IMAGE_ID")
	if _, err := client.GetArtifact(slugRunnerID); err != nil {
		return fmt.Errorf("Error getting slugrunner image: %s", err)
	}

	app, err := client.GetApp(appName)
	if err == controller.ErrNotFound {
		return fmt.Errorf("Unknown app %q", appName)
	} else if err != nil {
		return fmt.Errorf("Error retrieving app: %s", err)
	}
	prevRelease, err := client.GetAppRelease(app.Name)
	if err == controller.ErrNotFound {
		prevRelease = &ct.Release{}
	} else if err != nil {
		return fmt.Errorf("Error getting current app release: %s", err)
	}

	fmt.Printf("-----> Building %s...\n", app.Name)

	slugImageID := random.UUID()
	jobEnv := map[string]string{
		"BUILD_CACHE_URL": fmt.Sprintf("%s/%s-cache.tgz", blobstoreURL, app.ID),
		"CONTROLLER_KEY":  os.Getenv("CONTROLLER_KEY"),
		"SLUG_IMAGE_ID":   slugImageID,
	}
	if buildpackURL, ok := env["BUILDPACK_URL"]; ok {
		jobEnv["BUILDPACK_URL"] = buildpackURL
	} else if buildpackURL, ok := prevRelease.Env["BUILDPACK_URL"]; ok {
		jobEnv["BUILDPACK_URL"] = buildpackURL
	}
	for _, k := range []string{"SSH_CLIENT_KEY", "SSH_CLIENT_HOSTS"} {
		if v := os.Getenv(k); v != "" {
			jobEnv[k] = v
		}
	}

	job := &host.Job{
		Config: host.ContainerConfig{
			Args:       []string{"/builder/build.sh"},
			Env:        jobEnv,
			Stdin:      true,
			DisableLog: true,
		},
		Partition: "background",
		Metadata: map[string]string{
			"flynn-controller.app":      app.ID,
			"flynn-controller.app_name": app.Name,
			"flynn-controller.release":  prevRelease.ID,
			"flynn-controller.type":     "slugbuilder",
		},
		Resources: resource.Defaults(),
	}
	if sb, ok := prevRelease.Processes["slugbuilder"]; ok {
		job.Resources = sb.Resources
	} else if rawLimit := os.Getenv("SLUGBUILDER_DEFAULT_MEMORY_LIMIT"); rawLimit != "" {
		if limit, err := resource.ParseLimit(resource.TypeMemory, rawLimit); err == nil {
			job.Resources[resource.TypeMemory] = resource.Spec{Limit: &limit, Request: &limit}
		}
	}

	cmd := exec.Job(slugBuilder, job)
	cmd.Volumes = []*ct.VolumeReq{{Path: "/tmp", DeleteOnStop: true}}
	var output bytes.Buffer
	cmd.Stdout = io.MultiWriter(os.Stdout, &output)
	cmd.Stderr = os.Stderr

	releaseEnv := make(map[string]string, len(env))
	if prevRelease.Env != nil {
		for k, v := range prevRelease.Env {
//.........这里部分代码省略.........

func (s *CLISuite) TestExportImport(t *c.C) {
	srcApp := "app-export" + random.String(8)
	dstApp := "app-import" + random.String(8)

	// create app
	r := s.newGitRepo(t, "http")
	t.Assert(r.flynn("create", srcApp), Succeeds)

	// exporting the app without a release should work
	file := filepath.Join(t.MkDir(), "export.tar")
	t.Assert(r.flynn("export", "-f", file), Succeeds)
	assertExportContains(t, file, "app.json", "routes.json")

	// exporting the app with an artifact-less release should work
	t.Assert(r.flynn("env", "set", "FOO=BAR"), Succeeds)
	t.Assert(r.flynn("export", "-f", file), Succeeds)
	assertExportContains(t, file, "app.json", "routes.json", "release.json")

	// release the app and provision some dbs
	t.Assert(r.git("push", "flynn", "master"), Succeeds)
	t.Assert(r.flynn("resource", "add", "postgres"), Succeeds)
	t.Assert(r.flynn("pg", "psql", "--", "-c",
		"CREATE table foos (data text); INSERT INTO foos (data) VALUES ('foobar')"), Succeeds)
	t.Assert(r.flynn("resource", "add", "mysql"), Succeeds)
	t.Assert(r.flynn("mysql", "console", "--", "-e",
		"CREATE TABLE foos (data TEXT); INSERT INTO foos (data) VALUES ('foobar')"), Succeeds)

	// grab the slug details
	client := s.controllerClient(t)
	release, err := client.GetAppRelease(srcApp)
	t.Assert(err, c.IsNil)
	artifact, err := client.GetArtifact(release.ArtifactIDs[1])
	t.Assert(err, c.IsNil)
	slugLayer := artifact.Manifest().Rootfs[0].Layers[0]

	// export app
	t.Assert(r.flynn("export", "-f", file), Succeeds)
	assertExportContains(t, file,
		"app.json", "routes.json", "release.json", "artifacts.json",
		slugLayer.ID+".layer", "formation.json",
		"postgres.dump", "mysql.dump",
	)

	// remove db tables from source app
	t.Assert(r.flynn("pg", "psql", "--", "-c", "DROP TABLE foos"), Succeeds)
	t.Assert(r.flynn("mysql", "console", "--", "-e", "DROP TABLE foos"), Succeeds)

	// remove the git remote
	t.Assert(r.git("remote", "remove", "flynn"), Succeeds)

	// import app
	t.Assert(r.flynn("import", "--name", dstApp, "--file", file), Succeeds)

	// test dbs were imported
	query := r.flynn("-a", dstApp, "pg", "psql", "--", "-c", "SELECT * FROM foos")
	t.Assert(query, SuccessfulOutputContains, "foobar")
	query = r.flynn("-a", dstApp, "mysql", "console", "--", "-e", "SELECT * FROM foos")
	t.Assert(query, SuccessfulOutputContains, "foobar")

	// wait for it to start
	_, err = s.discoverdClient(t).Instances(dstApp+"-web", 10*time.Second)
	t.Assert(err, c.IsNil)
}

//.........这里部分代码省略.........
		}.RunWithState(ch, state)
		if err != nil {
			return err
		}

		// ensure the formation is correct in the database
		if err := client.PutFormation(data.MongoDB.Formation()); err != nil {
			return fmt.Errorf("error updating mongodb formation: %s", err)
		}

		cmd = exec.JobUsingHost(state.Hosts[0], artifacts["mongodb"], nil)
		cmd.Args = []string{"mongorestore", "-h", "leader.mongodb.discoverd", "-u", "flynn", "-p", data.MongoDB.Release.Env["MONGO_PWD"], "--archive"}
		cmd.Stdin = mongodb
		meta = bootstrap.StepMeta{ID: "restore", Action: "restore-mongodb"}
		ch <- &bootstrap.StepInfo{StepMeta: meta, State: "start", Timestamp: time.Now().UTC()}
		out, err = cmd.CombinedOutput()
		if os.Getenv("DEBUG") != "" {
			fmt.Println(string(out))
		}
		if err != nil {
			ch <- &bootstrap.StepInfo{
				StepMeta:  meta,
				State:     "error",
				Error:     fmt.Sprintf("error running mongodb restore: %s - %q", err, string(out)),
				Err:       err,
				Timestamp: time.Now().UTC(),
			}
			return err
		}
		ch <- &bootstrap.StepInfo{StepMeta: meta, State: "done", Timestamp: time.Now().UTC()}
	}

	// get blobstore config
	blobstoreRelease, err := client.GetAppRelease("blobstore")
	if err != nil {
		return fmt.Errorf("error getting blobstore release: %s", err)
	}
	blobstoreFormation, err := client.GetExpandedFormation("blobstore", blobstoreRelease.ID)
	if err != nil {
		return fmt.Errorf("error getting blobstore expanded formation: %s", err)
	}
	state.SetControllerKey(controllerKey)
	ch <- &bootstrap.StepInfo{StepMeta: meta, State: "done", Timestamp: time.Now().UTC()}

	// start the blobstore
	blobstoreFormation.Artifacts = []*ct.Artifact{artifacts["blobstore"]}
	_, err = bootstrap.Manifest{
		step("blobstore", "run-app", &bootstrap.RunAppAction{
			ExpandedFormation: blobstoreFormation,
		}),
		step("blobstore-wait", "wait", &bootstrap.WaitAction{
			URL:    "http://blobstore.discoverd",
			Status: 200,
		}),
	}.RunWithState(ch, state)
	if err != nil {
		return err
	}

	// now that the controller and blobstore are up and controller
	// migrations have run (so we know artifacts have a manifest column),
	// migrate all artifacts to Flynn images
	jsonb := func(v interface{}) []byte {
		data, _ := json.Marshal(v)
		return data
	}

//.........这里部分代码省略.........
	}
	logR, logW := io.Pipe()
	defer logW.Close()
	go func() {
		buf := bufio.NewReader(logR)
		for {
			line, err := buf.ReadString('\n')
			if err != nil {
				return
			}
			debug(t, line[0:len(line)-1])
		}
	}()
	cmd.Stdout = logW
	cmd.Stderr = logW
	t.Assert(cmd.Run(), c.IsNil)

	debug(t, "waiting for nodejs-web service")
	disc := discoverd.NewClientWithURL(fmt.Sprintf("http://%s:1111", ip))
	_, err = disc.Instances("nodejs-web", 30*time.Second)
	t.Assert(err, c.IsNil)

	debug(t, "checking HTTP requests")
	req, err := http.NewRequest("GET", "http://"+ip, nil)
	t.Assert(err, c.IsNil)
	req.Host = "nodejs.1.localflynn.com"
	var res *http.Response
	// try multiple times in case we get a 503 from the router as it has
	// not seen the service yet
	err = attempt.Strategy{Total: 10 * time.Second, Delay: 100 * time.Millisecond}.Run(func() (err error) {
		res, err = http.DefaultClient.Do(req)
		if err != nil {
			return err
		} else if res.StatusCode == http.StatusServiceUnavailable {
			return errors.New("router returned 503")
		}
		return nil
	})
	t.Assert(err, c.IsNil)
	t.Assert(res.StatusCode, c.Equals, http.StatusOK)

	debug(t, "getting app release")
	controllerInstances, err := disc.Instances("controller", 30*time.Second)
	t.Assert(err, c.IsNil)
	controllerURL := "http://" + controllerInstances[0].Addr
	controllerKey := controllerInstances[0].Meta["AUTH_KEY"]
	client, err := controller.NewClient(controllerURL, controllerKey)
	t.Assert(err, c.IsNil)
	release, err := client.GetAppRelease("nodejs")
	t.Assert(err, c.IsNil)

	debug(t, "configuring flynn CLI")
	flynnrc := filepath.Join(dir, ".flynnrc")
	conf := &config.Config{}
	t.Assert(conf.Add(&config.Cluster{
		Name:          "default",
		ControllerURL: controllerURL,
		Key:           controllerKey,
	}, true), c.IsNil)
	t.Assert(conf.SaveTo(flynnrc), c.IsNil)
	flynn := func(cmdArgs ...string) *CmdResult {
		cmd := exec.Command(args.CLI, cmdArgs...)
		cmd.Env = flynnEnv(flynnrc)
		cmd.Env = append(cmd.Env, "FLYNN_APP=nodejs")
		return run(t, cmd)
	}

	if _, ok := release.Env["FLYNN_REDIS"]; ok {
		debug(t, "checking redis resource")
		// try multiple times as the Redis resource is not guaranteed to be up yet
		var redisResult *CmdResult
		err = attempt.Strategy{Total: 10 * time.Second, Delay: 100 * time.Millisecond}.Run(func() error {
			redisResult = flynn("redis", "redis-cli", "--", "PING")
			return redisResult.Err
		})
		t.Assert(err, c.IsNil)
		t.Assert(redisResult, SuccessfulOutputContains, "PONG")
	}

	debug(t, "checking mysql resource")
	if _, ok := release.Env["FLYNN_MYSQL"]; ok {
		t.Assert(flynn("mysql", "console", "--", "-e", "SELECT * FROM foos"), SuccessfulOutputContains, "foobar")
	} else {
		t.Assert(flynn("resource", "add", "mysql"), Succeeds)
	}

	debug(t, "checking mongodb resource")
	if _, ok := release.Env["FLYNN_MONGO"]; ok {
		t.Assert(flynn("mongodb", "mongo", "--", "--eval", "db.foos.find()"), SuccessfulOutputContains, "foobar")
	} else {
		t.Assert(flynn("resource", "add", "mongodb"), Succeeds)
	}

	debug(t, "checking dashboard STATUS_KEY matches status AUTH_KEY")
	dashboardStatusKeyResult := flynn("-a", "dashboard", "env", "get", "STATUS_KEY")
	t.Assert(dashboardStatusKeyResult, Succeeds)
	statusAuthKeyResult := flynn("-a", "status", "env", "get", "AUTH_KEY")
	t.Assert(statusAuthKeyResult, Succeeds)
	t.Assert(dashboardStatusKeyResult.Output, c.Equals, statusAuthKeyResult.Output)
}

//.........这里部分代码省略.........
			return err
		}
		ch <- &bootstrap.StepInfo{StepMeta: meta, State: "done", Timestamp: time.Now().UTC()}
	}

	var mongodb io.Reader
	if data.MongoDB != nil {
		mongodb, err = getFile("mongodb.archive.gz")
		if err != nil {
			return err
		}
	}

	// load data into mongodb if it was present in the backup.
	if mongodb != nil && data.MongoDB != nil {
		cmd = exec.JobUsingHost(state.Hosts[0], host.Artifact{Type: data.MongoDB.ImageArtifact.Type, URI: data.MongoDB.ImageArtifact.URI}, nil)
		cmd.Args = []string{"mongorestore", "-h", "leader.mongodb.discoverd", "-u", "flynn", "-p", data.MongoDB.Release.Env["MONGO_PWD"], "--archive"}
		cmd.Stdin = mongodb
		meta = bootstrap.StepMeta{ID: "restore", Action: "restore-mongodb"}
		ch <- &bootstrap.StepInfo{StepMeta: meta, State: "start", Timestamp: time.Now().UTC()}
		out, err = cmd.CombinedOutput()
		if os.Getenv("DEBUG") != "" {
			fmt.Println(string(out))
		}
		if err != nil {
			ch <- &bootstrap.StepInfo{
				StepMeta:  meta,
				State:     "error",
				Error:     fmt.Sprintf("error running mongodb restore: %s - %q", err, string(out)),
				Err:       err,
				Timestamp: time.Now().UTC(),
			}
			return err
		}
		ch <- &bootstrap.StepInfo{StepMeta: meta, State: "done", Timestamp: time.Now().UTC()}
	}

	// start controller API
	data.Controller.Processes = map[string]int{"web": 1}
	_, err = bootstrap.Manifest{
		step("controller", "run-app", &bootstrap.RunAppAction{
			ExpandedFormation: data.Controller,
		}),
	}.RunWithState(ch, state)

	// wait for controller to come up
	meta = bootstrap.StepMeta{ID: "wait-controller", Action: "wait-controller"}
	ch <- &bootstrap.StepInfo{StepMeta: meta, State: "start", Timestamp: time.Now().UTC()}
	controllerInstances, err := discoverd.GetInstances("controller", 30*time.Second)
	if err != nil {
		return fmt.Errorf("error getting controller instance: %s", err)
	}

	// get blobstore config
	client, err := controller.NewClient("http://"+controllerInstances[0].Addr, data.Controller.Release.Env["AUTH_KEY"])
	if err != nil {
		return err
	}
	blobstoreRelease, err := client.GetAppRelease("blobstore")
	if err != nil {
		return fmt.Errorf("error getting blobstore release: %s", err)
	}
	blobstoreFormation, err := client.GetExpandedFormation("blobstore", blobstoreRelease.ID)
	if err != nil {
		return fmt.Errorf("error getting blobstore expanded formation: %s", err)
	}
	state.SetControllerKey(data.Controller.Release.Env["AUTH_KEY"])
	ch <- &bootstrap.StepInfo{StepMeta: meta, State: "done", Timestamp: time.Now().UTC()}

	// start blobstore, scheduler, and enable cluster monitor
	data.Controller.Processes = map[string]int{"scheduler": 1}
	// only start one scheduler instance
	schedulerProcess := data.Controller.Release.Processes["scheduler"]
	schedulerProcess.Omni = false
	data.Controller.Release.Processes["scheduler"] = schedulerProcess
	_, err = bootstrap.Manifest{
		step("blobstore", "run-app", &bootstrap.RunAppAction{
			ExpandedFormation: blobstoreFormation,
		}),
		step("blobstore-wait", "wait", &bootstrap.WaitAction{
			URL:    "http://blobstore.discoverd",
			Status: 200,
		}),
		step("controller-scheduler", "run-app", &bootstrap.RunAppAction{
			ExpandedFormation: data.Controlle