func TestSetAndGetStatus(t *testing.T) {
	store := newFixture()

	serviceStatus := ServiceStatus{
		Name:     "echo_service",
		State:    Running,
		LastExit: nil,
	}
	testStatus := PodStatus{
		ServiceStatus: []ServiceStatus{
			serviceStatus,
		},
	}

	podKey := types.NewPodUUID()
	err := store.Set(podKey, testStatus)
	if err != nil {
		t.Fatalf("Unexpected error setting status: %s", err)
	}

	status, _, err := store.Get(podKey)
	if err != nil {
		t.Fatalf("Unexpected error getting status: %s", err)
	}

	if len(status.ServiceStatus) != 1 {
		t.Fatalf("Expected one service status entry, but there were %d", len(status.ServiceStatus))
	}

	if status.ServiceStatus[0] != serviceStatus {
		t.Errorf("Status entry expected to be '%+v', was %+v", serviceStatus, status.ServiceStatus[0])
	}
}

func TestSetAndGetStatus(t *testing.T) {
	store := newFixture()

	processStatus := ProcessStatus{
		EntryPoint:   "echo_service",
		LaunchableID: "some_launchable",
		LastExit:     nil,
	}
	testStatus := PodStatus{
		ProcessStatuses: []ProcessStatus{
			processStatus,
		},
	}

	podKey := types.NewPodUUID()
	err := store.Set(podKey, testStatus)
	if err != nil {
		t.Fatalf("Unexpected error setting status: %s", err)
	}

	status, _, err := store.Get(podKey)
	if err != nil {
		t.Fatalf("Unexpected error getting status: %s", err)
	}

	if len(status.ProcessStatuses) != 1 {
		t.Fatalf("Expected one service status entry, but there were %d", len(status.ProcessStatuses))
	}

	if status.ProcessStatuses[0] != processStatus {
		t.Errorf("Status entry expected to be '%+v', was %+v", processStatus, status.ProcessStatuses[0])
	}
}

func TestWriteRealityIndex(t *testing.T) {
	node := types.NodeName("some_node")
	key := types.NewPodUUID()

	realityIndexPath := fmt.Sprintf("reality/%s/%s", node, key.ID)

	store, fakeKV := storeWithFakeKV(t, make(map[string]Pod), make(map[string]PodIndex))

	// confirm that the reality index doesn't exist
	pair, _, err := fakeKV.Get(realityIndexPath, nil)
	if err != nil {
		t.Fatalf("Initial conditions were not met: error fetching %s: %s", realityIndexPath, err)
	}

	if pair != nil {
		t.Fatalf("Initial conditions were not met: expected key %s to not exist", realityIndexPath)
	}

	err = store.WriteRealityIndex(key, node)
	if err != nil {
		t.Fatal(err)
	}

	pair, _, err = fakeKV.Get(realityIndexPath, nil)
	if err != nil {
		t.Fatalf("Unable to fetch the deleted key (%s): %s", realityIndexPath, err)
	}

	if pair == nil {
		t.Fatalf("%s should have been written but it wasn't", realityIndexPath)
	}
}

// UpdatePods looks at the pods currently being monitored and
// compares that to what the reality store indicates should be
// running. UpdatePods then shuts down the monitors for dead
// pods and creates PodWatch structs for new pods.
func TestUpdatePods(t *testing.T) {
	var current []PodWatch
	var reality []kp.ManifestResult
	// ids for current: 0, 1, 2, 3
	for i := 0; i < 4; i++ {
		current = append(current, *newWatch(types.PodID(strconv.Itoa(i))))
	}
	// ids for reality: 1, 2, test
	for i := 1; i < 3; i++ {
		// Health checking is not supported for uuid pods, so ensure that even
		// if /reality contains a uuid pod we don't actually watch its health
		uuidKeyResult := newManifestResult("some_uuid_pod")
		uuidKeyResult.PodUniqueKey = types.NewPodUUID()
		reality = append(reality, uuidKeyResult)
		reality = append(reality, newManifestResult(current[i].manifest.ID()))
	}
	reality = append(reality, newManifestResult("test"))

	// ids for pods: 1, 2, test
	// 0, 3 should have values in their shutdownCh
	logger := logging.NewLogger(logrus.Fields{})
	pods := updatePods(&MockHealthManager{}, nil, nil, current, reality, "", &logger)
	Assert(t).AreEqual(true, <-current[0].shutdownCh, "this PodWatch should have been shutdown")
	Assert(t).AreEqual(true, <-current[3].shutdownCh, "this PodWatch should have been shutdown")

	Assert(t).AreEqual(current[1].manifest.ID(), pods[0].manifest.ID(), "pod with id:1 should have been returned")
	Assert(t).AreEqual(current[2].manifest.ID(), pods[1].manifest.ID(), "pod with id:1 should have been returned")
	Assert(t).AreEqual("test", string(pods[2].manifest.ID()), "should have added pod with id:test to list")
}

func TestRun(t *testing.T) {
	tempDir, err := ioutil.TempDir("", "process_reporter")
	if err != nil {
		t.Fatalf("Could not create temp dir: %s", err)
	}
	defer os.RemoveAll(tempDir)

	dbPath, quitCh, podStatusStore := startReporter(t, tempDir)
	defer close(quitCh)

	finishOutput1 := FinishOutput{
		PodID:        "some_pod",
		LaunchableID: "some_launchable",
		EntryPoint:   "launch",
		PodUniqueKey: types.NewPodUUID(),
		ExitCode:     1,
		ExitStatus:   120,
	}

	finishService, err := NewSQLiteFinishService(dbPath, logging.DefaultLogger)
	if err != nil {
		t.Fatalf("Could not initialize finish service: %s", err)
	}
	defer finishService.Close()

	err = finishService.Insert(finishOutput1)
	if err != nil {
		t.Fatalf("Could not insert first finish value into the database: %s", err)
	}

	assertStatusUpdated(t, finishOutput1, podStatusStore)

	finishOutput2 := FinishOutput{
		PodID:        "some_pod",
		LaunchableID: "some_launchable",
		EntryPoint:   "nginx_worker",
		PodUniqueKey: types.NewPodUUID(),
		ExitCode:     3,
		ExitStatus:   67,
	}
	err = finishService.Insert(finishOutput2)
	if err != nil {
		t.Fatalf("Could not insert second finish value into the database: %s", err)
	}

	assertStatusUpdated(t, finishOutput2, podStatusStore)
}

func TestPodUniqueKeyFromConsulPath(t *testing.T) {
	type expectation struct {
		path string
		err  bool
		uuid types.PodUniqueKey
	}

	uuid := types.NewPodUUID()
	expectations := []expectation{
		{
			path: "intent/example.com/mysql",
			err:  false,
			uuid: "",
		},
		{
			path: "reality/example.com/mysql",
			err:  false,
			uuid: "",
		},
		{
			path: "labels/example.com/mysql",
			err:  true,
		},
		{
			path: "intent/example/com/mysql",
			err:  true,
		},
		{
			path: "hooks/all_hooks",
			err:  false,
			uuid: "",
		},
		{
			path: fmt.Sprintf("intent/example.com/%s", uuid),
			uuid: uuid,
			err:  false,
		},
	}

	for _, expectation := range expectations {
		podUniqueKey, err := PodUniqueKeyFromConsulPath(expectation.path)
		if expectation.err {
			if err == nil {
				t.Errorf("Expected an error for key '%s'", expectation.path)
			}
			continue
		}

		if err != nil {
			t.Errorf("Unexpected error for key '%s': %s", expectation.path, err)
			continue
		}

		if podUniqueKey != expectation.uuid {
			t.Errorf("Expected podUniqueKey to be %s, was %s", expectation.uuid, podUniqueKey)
		}
	}
}

func TestPruneRowsBefore(t *testing.T) {
	finishService, dbPath, closeFunc := initFinishService(t)
	defer closeFunc()
	defer finishService.Close()

	// Insert 3 rows
	for i := 0; i < 3; i++ {
		// now check that we can insert a finish to confirm finishes table was created
		err := finishService.Insert(FinishOutput{
			PodID:        "some_pod",
			PodUniqueKey: types.NewPodUUID(),
			LaunchableID: "some_launchable",
			EntryPoint:   "launch",
			ExitCode:     4,
			ExitStatus:   127,
		})
		if err != nil {
			t.Errorf("Could not insert a finish row, the finishes table might not have been created: %s", err)
		}
	}

	// Open up the DB directly so we can mess with the dates
	db, err := sql.Open("sqlite3", dbPath)
	if err != nil {
		t.Fatalf("Could not open database to confirm migrations worked: %s", err)
	}
	defer db.Close()

	// Set the dates of the first two rows to a week in the past
	_, err = db.Exec(`
	update finishes
	set date = ?
	where id < 3
	`, time.Now().Add(-7*24*time.Hour))
	if err != nil {
		t.Fatalf("Could not set time back a week for two rows: %s", err)
	}

	// Now prune any rows older than an hour, we should see the first two deleted
	err = finishService.PruneRowsBefore(time.Now().Add(-time.Hour))
	if err != nil {
		t.Fatalf("Unexpected error pruning rows: %s", err)
	}

	finishes, err := finishService.GetLatestFinishes(0)
	if err != nil {
		t.Fatalf("Could not fetch all finishes from the table to verify pruning: %s", err)
	}

	if len(finishes) != 1 {
		t.Fatalf("Expected only 1 row after pruning, but there were %d", len(finishes))
	}
}

func TestReadPodFromIndex(t *testing.T) {
	node := types.NodeName("some_node")
	key := types.NewPodUUID()

	podPath := fmt.Sprintf("pods/%s", key.ID)
	indexPath := fmt.Sprintf("intent/%s/%s", node, key.ID)

	pod := Pod{
		Manifest: testManifest(),
		Node:     node,
	}

	index := PodIndex{
		PodKey: key,
	}

	// Initialize the store with entries at the pod path and index path
	pods := map[string]Pod{
		podPath: pod,
	}

	// This test doesn't actually care if there's an index, but let's keep it hygienic
	indices := map[string]PodIndex{
		indexPath: index,
	}
	store, _ := storeWithFakeKV(t, pods, indices)

	outPod, err := store.ReadPod(key)
	if err != nil {
		t.Fatalf("Unexpected error reading pod: %s", err)
	}

	if pod.Node != outPod.Node {
		t.Errorf("Pod node, didn't match expected, wanted %+v was %+v", pod.Node, outPod.Node)
	}

	expectedManifestSha, err := testManifest().SHA()
	if err != nil {
		t.Fatal(err)
	}

	podManifestSha, err := outPod.Manifest.SHA()
	if err != nil {
		t.Fatal(err)
	}

	if expectedManifestSha != podManifestSha {
		t.Error("Pod returned with the wrong manifest")
	}
}

func TestMutateStatusExistingKey(t *testing.T) {
	store := newFixture()

	key := types.NewPodUUID()
	processStatus := ProcessStatus{
		EntryPoint:   "echo_service",
		LaunchableID: "some_launchable",
		LastExit:     nil,
	}
	err := store.Set(key, PodStatus{
		ProcessStatuses: []ProcessStatus{
			processStatus,
		},
	})
	if err != nil {
		t.Fatalf("Unable to set up test with an existing key: %s", err)
	}

	err = store.MutateStatus(key, func(p PodStatus) (PodStatus, error) {
		p.PodStatus = PodLaunched
		return p, nil
	})
	if err != nil {
		t.Fatal(err)
	}

	// Now try to get it and confirm the status was set
	status, _, err := store.Get(key)
	if err != nil {
		t.Fatal(err)
	}

	if status.PodStatus != PodLaunched {
		t.Errorf("Expected pod status to be set to '%s' but was '%s'", PodLaunched, status.PodStatus)
	}

	if len(status.ProcessStatuses) != 1 {
		t.Error("ProcessStatus field didn't go untouched when mutating PodStatus")
	}
}

func TestMutateStatusNewKey(t *testing.T) {
	store := newFixture()

	key := types.NewPodUUID()
	err := store.MutateStatus(key, func(p PodStatus) (PodStatus, error) {
		p.PodStatus = PodLaunched
		return p, nil
	})
	if err != nil {
		t.Fatal(err)
	}

	// Now try to get it and confirm the status was set
	status, _, err := store.Get(key)
	if err != nil {
		t.Fatal(err)
	}

	if status.PodStatus != PodLaunched {
		t.Errorf("Expected pod status to be set to '%s' but was '%s'", PodLaunched, status.PodStatus)
	}
}

func TestMigrate(t *testing.T) {
	finishService, dbPath, closeFunc := initFinishService(t)
	defer closeFunc()
	defer finishService.Close()

	// now check that we can insert a finish to confirm finishes table was created
	err := finishService.Insert(FinishOutput{
		PodID:        "some_pod",
		PodUniqueKey: types.NewPodUUID(),
		LaunchableID: "some_launchable",
		EntryPoint:   "launch",
		ExitCode:     4,
		ExitStatus:   127,
	})
	if err != nil {
		t.Errorf("Could not insert a finish row, the finishes table might not have been created: %s", err)
	}

	db, err := sql.Open("sqlite3", dbPath)
	if err != nil {
		t.Fatalf("Could not open database to confirm migrations worked: %s", err)
	}
	defer db.Close()

	// now check schema version
	var schemaVersion int64
	err = db.QueryRow(getSchemaVersionQuery).Scan(&schemaVersion)
	switch {
	case err == sql.ErrNoRows:
		t.Fatal("Schema version table was not written")
	case err != nil:
		t.Fatalf("Could not check schema version: %s", err)
	}

	if schemaVersion != int64(len(sqliteMigrations)) {
		t.Errorf("Expected schema_version to be %d but was %d", len(sqliteMigrations), schemaVersion)
	}
}

func TestUnschedule(t *testing.T) {
	node := types.NodeName("some_node")
	key := types.NewPodUUID()

	podPath := fmt.Sprintf("pods/%s", key.ID)
	indexPath := fmt.Sprintf("intent/%s/%s", node, key.ID)

	// Initialize the store with entries at the pod path and index path
	pods := map[string]Pod{
		podPath: {
			Manifest: testManifest(),
			Node:     node,
		},
	}

	indices := map[string]PodIndex{
		indexPath: {
			PodKey: key,
		},
	}
	store, kv := storeWithFakeKV(t, pods, indices)

	// Now delete the pod entry
	err := store.Unschedule(key)
	if err != nil {
		t.Fatalf("Unexpected error deleting pod: %s", err)
	}

	if kv.Entries[podPath] != nil {
		t.Fatalf("Key '%s' was deleted as expected", podPath)
	}

	if kv.Entries[indexPath] != nil {
		t.Fatalf("Index '%s' was deleted as expected", indexPath)
	}
}

func TestWatchPodStatus(t *testing.T) {
	respCh := make(chan *podstore_protos.PodStatusResponse)
	stream := &WatchPodStatusStream{
		FakeServerStream: testutil.NewFakeServerStream(context.Background()),
		ResponseCh:       respCh,
	}

	podStatusStore, server := setupServerWithFakePodStatusStore()

	podUniqueKey := types.NewPodUUID()
	req := &podstore_protos.WatchPodStatusRequest{
		StatusNamespace: kp.PreparerPodStatusNamespace.String(),
		PodUniqueKey:    podUniqueKey.String(),
		// Set it 1 above last index so we wait for the key to exist. (The test status
		// store starts at 1234 for some reason)
		WaitIndex: 1235,
	}

	watchErrCh := make(chan error)
	defer close(watchErrCh)
	go func() {
		err := server.WatchPodStatus(req, stream)
		if err != nil {
			watchErrCh <- err
		}
	}()

	expectedTime := time.Now()
	expectedManifest := `id: "test_app"`
	expectedPodState := podstatus.PodLaunched
	expectedLaunchableID := launch.LaunchableID("nginx")
	expectedEntryPoint := "launch"
	expectedExitCode := 3
	expectedExitStatus := 4
	setStatusErrCh := make(chan error)
	defer close(setStatusErrCh)
	go func() {
		err := podStatusStore.Set(podUniqueKey, podstatus.PodStatus{
			Manifest:  expectedManifest,
			PodStatus: expectedPodState,
			ProcessStatuses: []podstatus.ProcessStatus{
				{
					LaunchableID: expectedLaunchableID,
					EntryPoint:   expectedEntryPoint,
					LastExit: &podstatus.ExitStatus{
						ExitTime:   expectedTime,
						ExitCode:   expectedExitCode,
						ExitStatus: expectedExitStatus,
					},
				},
			},
		})
		if err != nil {
			setStatusErrCh <- err
		}
	}()

	select {
	case <-time.After(5 * time.Second):
		t.Fatal("Didn't receive value after 5 seconds")
	case err := <-setStatusErrCh:
		t.Fatalf("Error setting status to trigger watch: %s", err)
	case err := <-watchErrCh:
		t.Fatalf("Unexpected error watching for status: %s", err)
	case resp := <-respCh:
		if resp.Manifest != expectedManifest {
			t.Errorf("Manifest didn't match expected, wanted %q got %q", expectedManifest, resp.Manifest)
		}

		if resp.PodState != expectedPodState.String() {
			t.Errorf("PodState didn't match expcted, wanted %q got %q", expectedPodState, resp.PodState)
		}

		if len(resp.ProcessStatuses) != 1 {
			t.Fatalf("Expected 1 process status in pod status but got %d", len(resp.ProcessStatuses))
		}

		processStatus := resp.ProcessStatuses[0]
		if processStatus.LaunchableId != expectedLaunchableID.String() {
			t.Errorf("Expected process status for launchable %q but found %q", expectedLaunchableID, processStatus.LaunchableId)
		}

		if processStatus.EntryPoint != expectedEntryPoint {
			t.Errorf("Expected process status for entry point %q but found %q", expectedEntryPoint, processStatus.EntryPoint)
		}

		if processStatus.LastExit == nil {
			t.Fatal("Expected exit information for process")
		}

		lastExit := processStatus.LastExit
		if lastExit.ExitTime != expectedTime.Unix() {
			t.Error("Exit time for process in status didn't match expected")
		}

		if lastExit.ExitCode != int64(expectedExitCode) {
			t.Errorf("Expected exit code %d but got %d", expectedExitCode, lastExit.ExitCode)
		}

		if lastExit.ExitStatus != int64(expectedExitStatus) {
//.........这里部分代码省略.........

func (c *consulStore) Schedule(manifest manifest.Manifest, node types.NodeName) (key types.PodUniqueKey, err error) {
	manifestBytes, err := manifest.Marshal()
	if err != nil {
		return "", err
	}

	podKey := types.NewPodUUID()

	podPath := computePodPath(podKey)
	intentIndexPath := computeIntentIndexPath(podKey, node)

	// Write the Pod to /pods/<key>
	pod := RawPod{
		Manifest: string(manifestBytes),
		Node:     node,
	}

	podBytes, err := json.Marshal(pod)
	if err != nil {
		return "", err
	}

	pair := &api.KVPair{
		Key:   podPath,
		Value: podBytes,
	}

	_, err = c.consulKV.Put(pair, nil)
	if err != nil {
		return "", consulutil.NewKVError("put", podPath, err)
	}

	// Now, write the secondary index to /intent/<node>/<key>
	index := PodIndex{
		PodKey: podKey,
	}

	// NOTE: errors might happen after this point which means we've written
	// a pod to /pods and we haven't written the corresponding index. In
	// those cases, we do a single attempt to delete the main pod key. In
	// the event of a Consul outage it's likely that the cleanup will fail,
	// so there will be a pod with no secondary index. For that purpose, a
	// sweeper process is planned to remove pods for which there is no index.
	// TODO: use a transaction when we can rely upon consul 0.7
	defer func() {
		if err != nil {
			_, _ = c.consulKV.Delete(podPath, nil)
		}
	}()

	indexBytes, err := json.Marshal(index)
	if err != nil {
		return "", util.Errorf("Could not marshal index as json: %s", err)
	}

	indexPair := &api.KVPair{
		Key:   intentIndexPath,
		Value: indexBytes,
	}
	_, err = c.consulKV.Put(indexPair, nil)
	if err != nil {
		return "", consulutil.NewKVError("put", intentIndexPath, err)
	}

	return podKey, nil
}