// handleStoreStatus handles GET requests for a single node's status. If no id
// is available, it calls handleStoresStatus to return all store's statuses.
func (s *statusServer) handleStoreStatus(w http.ResponseWriter, r *http.Request, ps httprouter.Params) {
	id, err := strconv.ParseInt(ps.ByName("id"), 10, 32)
	if err != nil {
		log.Error(err)
		w.WriteHeader(http.StatusInternalServerError)
		return
	}
	key := keys.StoreStatusKey(int32(id))

	storeStatus := &storage.StoreStatus{}
	if err := s.db.GetProto(key, storeStatus); err != nil {
		log.Error(err)
		w.WriteHeader(http.StatusInternalServerError)
		return
	}

	b, contentType, err := util.MarshalResponse(r, storeStatus, []util.EncodingType{util.JSONEncoding})
	if err != nil {
		log.Error(err)
		w.WriteHeader(http.StatusInternalServerError)
		return
	}
	w.Header().Set(util.ContentTypeHeader, contentType)
	w.Write(b)
}

// writeSummaries retrieves status summaries from the supplied
// NodeStatusRecorder and persists them to the cockroach data store.
func (s *Server) writeSummaries() error {
	nodeStatus, storeStatuses := s.recorder.GetStatusSummaries()
	if nodeStatus != nil {
		key := keys.NodeStatusKey(int32(nodeStatus.Desc.NodeID))
		if err := s.db.Put(key, nodeStatus); err != nil {
			return err
		}
		if log.V(1) {
			statusJSON, err := json.Marshal(nodeStatus)
			if err != nil {
				log.Errorf("error marshaling nodeStatus to json: %s", err)
			}
			log.Infof("node %d status: %s", nodeStatus.Desc.NodeID, statusJSON)
		}
	}

	for _, ss := range storeStatuses {
		key := keys.StoreStatusKey(int32(ss.Desc.StoreID))
		if err := s.db.Put(key, &ss); err != nil {
			return err
		}
		if log.V(1) {
			statusJSON, err := json.Marshal(&ss)
			if err != nil {
				log.Errorf("error marshaling storeStatus to json: %s", err)
			}
			log.Infof("store %d status: %s", ss.Desc.StoreID, statusJSON)
		}
	}
	return nil
}

// writeSummaries retrieves status summaries from the supplied
// NodeStatusRecorder and persists them to the cockroach data store.
func (s *Server) writeSummaries() (err error) {
	s.stopper.RunTask(func() {
		nodeStatus, storeStatuses := s.recorder.GetStatusSummaries()
		if nodeStatus != nil {
			key := keys.NodeStatusKey(int32(nodeStatus.Desc.NodeID))
			if err = s.db.Put(key, nodeStatus); err != nil {
				return
			}
			if log.V(1) {
				log.Infof("recorded status for node %d", nodeStatus.Desc.NodeID)
			}
		}

		for _, ss := range storeStatuses {
			key := keys.StoreStatusKey(int32(ss.Desc.StoreID))
			if err = s.db.Put(key, &ss); err != nil {
				return
			}
		}
		if log.V(1) {
			log.Infof("recorded status for %d stores", len(storeStatuses))
		}
	})
	return nil
}

// writeSummaries retrieves status summaries from the supplied
// NodeStatusRecorder and persists them to the cockroach data store.
func (s *Server) writeSummaries() error {
	if !s.stopper.StartTask() {
		return nil
	}
	defer s.stopper.FinishTask()

	nodeStatus, storeStatuses := s.recorder.GetStatusSummaries()
	if nodeStatus != nil {
		key := keys.NodeStatusKey(int32(nodeStatus.Desc.NodeID))
		if err := s.db.Put(key, nodeStatus); err != nil {
			return err
		}
		if log.V(1) {
			log.Infof("recorded status for node %d", nodeStatus.Desc.NodeID)
		}
	}

	for _, ss := range storeStatuses {
		key := keys.StoreStatusKey(int32(ss.Desc.StoreID))
		if err := s.db.Put(key, &ss); err != nil {
			return err
		}
	}
	if log.V(1) {
		log.Infof("recorded status for %d stores", len(storeStatuses))
	}
	return nil
}

// compareStoreStatus ensures that the actual store status for the passed in
// store is updated correctly. It checks that the Desc.StoreID, Desc.Attrs,
// Desc.Node, Desc.Capacity.Capacity, NodeID, RangeCount, ReplicatedRangeCount
// are exactly correct and that the bytes and counts for Live, Key and Val are
// at least the expected value.
// The latest actual stats are returned.
func compareStoreStatus(t *testing.T, store *storage.Store, expectedStoreStatus *storage.StoreStatus, testNumber int) *storage.StoreStatus {
	storeStatusKey := keys.StoreStatusKey(int32(store.Ident.StoreID))
	gArgs, gReply := getArgs(storeStatusKey, 1, store.Ident.StoreID)
	if err := store.ExecuteCmd(context.Background(), proto.Call{Args: gArgs, Reply: gReply}); err != nil {
		t.Fatalf("%v: failure getting store status: %s", testNumber, err)
	}
	if gReply.Value == nil {
		t.Errorf("%v: could not find store status at: %s", testNumber, storeStatusKey)
	}
	storeStatus := &storage.StoreStatus{}
	if err := gogoproto.Unmarshal(gReply.Value.GetBytes(), storeStatus); err != nil {
		t.Fatalf("%v: could not unmarshal store status: %+v", testNumber, gReply)
	}

	// Values much match exactly.
	if expectedStoreStatus.Desc.StoreID != storeStatus.Desc.StoreID {
		t.Errorf("%v: actual Desc.StoreID does not match expected\nexpected: %+v\nactual: %v\n", testNumber, expectedStoreStatus, storeStatus)
	}
	if !reflect.DeepEqual(expectedStoreStatus.Desc.Attrs, storeStatus.Desc.Attrs) {
		t.Errorf("%v: actual Desc.Attrs does not match expected\nexpected: %+v\nactual: %v\n", testNumber, expectedStoreStatus, storeStatus)
	}
	if !reflect.DeepEqual(expectedStoreStatus.Desc.Node, storeStatus.Desc.Node) {
		t.Errorf("%v: actual Desc.Attrs does not match expected\nexpected: %+v\nactual: %v\n", testNumber, expectedStoreStatus, storeStatus)
	}
	if storeStatus.Desc.Capacity.Capacity != expectedStoreStatus.Desc.Capacity.Capacity {
		t.Errorf("%v: actual Desc.Capacity.Capacity does not match expected\nexpected: %+v\nactual: %v\n", testNumber, expectedStoreStatus, storeStatus)
	}
	if expectedStoreStatus.NodeID != storeStatus.NodeID {
		t.Errorf("%v: actual node ID does not match expected\nexpected: %+v\nactual: %v\n", testNumber, expectedStoreStatus, storeStatus)
	}
	if expectedStoreStatus.RangeCount != storeStatus.RangeCount {
		t.Errorf("%v: actual RangeCount does not match expected\nexpected: %+v\nactual: %v\n", testNumber, expectedStoreStatus, storeStatus)
	}
	if expectedStoreStatus.ReplicatedRangeCount != storeStatus.ReplicatedRangeCount {
		t.Errorf("%v: actual ReplicatedRangeCount does not match expected\nexpected: %+v\nactual: %v\n", testNumber, expectedStoreStatus, storeStatus)
	}

	// Values should be >= to expected values.
	if storeStatus.Stats.LiveBytes < expectedStoreStatus.Stats.LiveBytes {
		t.Errorf("%v: actual Live Bytes is not greater or equal to expected\nexpected: %+v\nactual: %v\n", testNumber, expectedStoreStatus, storeStatus)
	}
	if storeStatus.Stats.KeyBytes < expectedStoreStatus.Stats.KeyBytes {
		t.Errorf("%v: actual Key Bytes is not greater or equal to expected\nexpected: %+v\nactual: %v\n", testNumber, expectedStoreStatus, storeStatus)
	}
	if storeStatus.Stats.ValBytes < expectedStoreStatus.Stats.ValBytes {
		t.Errorf("%v: actual Val Bytes is not greater or equal to expected\nexpected: %+v\nactual: %v\n", testNumber, expectedStoreStatus, storeStatus)
	}
	if storeStatus.Stats.LiveCount < expectedStoreStatus.Stats.LiveCount {
		t.Errorf("%v: actual Live Count is not greater or equal to expected\nexpected: %+v\nactual: %v\n", testNumber, expectedStoreStatus, storeStatus)
	}
	if storeStatus.Stats.KeyCount < expectedStoreStatus.Stats.KeyCount {
		t.Errorf("%v: actual Key Count is not greater or equal to expected\nexpected: %+v\nactual: %v\n", testNumber, expectedStoreStatus, storeStatus)
	}
	if storeStatus.Stats.ValCount < expectedStoreStatus.Stats.ValCount {
		t.Errorf("%v: actual Val Count is not greater or equal to expected\nexpected: %+v\nactual: %v\n", testNumber, expectedStoreStatus, storeStatus)
	}
	return storeStatus
}

// compareStoreStatus ensures that the actual store status for the passed in
// store is updated correctly. It checks that the Desc.StoreID, Desc.Attrs,
// Desc.Node, Desc.Capacity.Capacity, NodeID, RangeCount, ReplicatedRangeCount
// are exactly correct and that the bytes and counts for Live, Key and Val are
// at least the expected value.
// The latest actual stats are returned.
func compareStoreStatus(t *testing.T, ts *TestServer, store *storage.Store, expectedStoreStatus *storage.StoreStatus, testNumber int) *storage.StoreStatus {
	// Retrieve store status from database.
	storeStatusKey := keys.StoreStatusKey(int32(store.Ident.StoreID))
	storeStatus := &storage.StoreStatus{}
	if err := ts.db.GetProto(storeStatusKey, storeStatus); err != nil {
		t.Fatalf("%v: failure getting store status: %s", testNumber, err)
	}

	// Values must match exactly.
	if a, e := storeStatus.Desc.StoreID, expectedStoreStatus.Desc.StoreID; a != e {
		t.Errorf("%d: actual Desc.StoreID does not match expected. expected: %d actual: %d", testNumber, e, a)
	}
	if a, e := storeStatus.Desc.Attrs, expectedStoreStatus.Desc.Attrs; !reflect.DeepEqual(a, e) {
		t.Errorf("%d: actual Desc.Attrs does not match expected.\nexpected: %s\nactual: %s", testNumber, e, a)
	}
	if a, e := storeStatus.Desc.Node, expectedStoreStatus.Desc.Node; !reflect.DeepEqual(a, e) {
		t.Errorf("%d: actual Desc.Attrs does not match expected.\nexpected: %s\nactual: %s", testNumber, e, a)
	}
	if a, e := storeStatus.Desc.Capacity.Capacity, expectedStoreStatus.Desc.Capacity.Capacity; a != e {
		t.Errorf("%d: actual Desc.Capacity.Capacity does not match expected.\nexpected: %d\nactual: %d", testNumber, e, a)
	}
	if a, e := storeStatus.NodeID, expectedStoreStatus.NodeID; a != e {
		t.Errorf("%d: actual node ID does not match expected.\nexpected: %d\nactual: %d", testNumber, e, a)
	}
	if a, e := storeStatus.RangeCount, expectedStoreStatus.RangeCount; a != e {
		t.Errorf("%d: actual RangeCount does not match expected.\nexpected: %d\nactual: %d", testNumber, e, a)
	}
	if a, e := storeStatus.ReplicatedRangeCount, expectedStoreStatus.ReplicatedRangeCount; a != e {
		t.Errorf("%d: actual ReplicatedRangeCount does not match expected.\nexpected: %d\nactual: %d", testNumber, e, a)
	}

	// Values should be >= to expected values.
	if a, e := storeStatus.Stats.LiveBytes, expectedStoreStatus.Stats.LiveBytes; a < e {
		t.Errorf("%d: actual Live Bytes is not greater or equal to expected.\nexpected: %d\nactual: %d", testNumber, e, a)
	}
	if a, e := storeStatus.Stats.KeyBytes, expectedStoreStatus.Stats.KeyBytes; a < e {
		t.Errorf("%d: actual Key Bytes is not greater or equal to expected.\nexpected: %d\nactual: %d", testNumber, e, a)
	}
	if a, e := storeStatus.Stats.ValBytes, expectedStoreStatus.Stats.ValBytes; a < e {
		t.Errorf("%d: actual Val Bytes is not greater or equal to expected.\nexpected: %d\nactual: %d", testNumber, e, a)
	}
	if a, e := storeStatus.Stats.LiveCount, expectedStoreStatus.Stats.LiveCount; a < e {
		t.Errorf("%d: actual Live Count is not greater or equal to expected.\nexpected: %d\nactual: %d", testNumber, e, a)
	}
	if a, e := storeStatus.Stats.KeyCount, expectedStoreStatus.Stats.KeyCount; a < e {
		t.Errorf("%d: actual Key Count is not greater or equal to expected.\nexpected: %d\nactual: %d", testNumber, e, a)
	}
	if a, e := storeStatus.Stats.ValCount, expectedStoreStatus.Stats.ValCount; a < e {
		t.Errorf("%d: actual Val Count is not greater or equal to expected.\nexpected: %d\nactual: %d", testNumber, e, a)
	}
	return storeStatus
}

// writeSummaries retrieves status summaries from the supplied
// NodeStatusRecorder and persists them to the cockroach data store.
func (n *Node) writeSummaries() error {
	var err error
	n.stopper.RunTask(func() {
		nodeStatus, storeStatuses := n.recorder.GetStatusSummaries()
		if nodeStatus != nil {
			key := keys.NodeStatusKey(int32(nodeStatus.Desc.NodeID))
			if pErr := n.ctx.DB.Put(key, nodeStatus); pErr != nil {
				err = pErr.GoError()
				return
			}
			if log.V(1) {
				statusJSON, err := json.Marshal(nodeStatus)
				if err != nil {
					log.Errorf("error marshaling nodeStatus to json: %s", err)
				}
				log.Infof("node %d status: %s", nodeStatus.Desc.NodeID, statusJSON)
			}
		}

		for _, ss := range storeStatuses {
			key := keys.StoreStatusKey(int32(ss.Desc.StoreID))
			if pErr := n.ctx.DB.Put(key, &ss); pErr != nil {
				err = pErr.GoError()
				return
			}
			if log.V(1) {
				statusJSON, err := json.Marshal(&ss)
				if err != nil {
					log.Errorf("error marshaling storeStatus to json: %s", err)
				}
				log.Infof("store %d status: %s", ss.Desc.StoreID, statusJSON)
			}
		}
	})
	return err
}