func (d *Datastore) Search(key ds.Key, query string) (*[]Model, error) {
	out, err := esCore.SearchRequest(true, d.esr.Index(key), key.Name(), query, "", 0)
	if err != nil {
		return nil, err
	}

	models := []Model{}
	for _, h := range out.Hits.Hits {

		// unmarshal response
		var data map[string]interface{}
		err := json.Unmarshal(h.Source, &data)
		if err != nil {
			return nil, fmt.Errorf("Unmarshal error: %v", err)
		}

		// get key
		k, ok := data["key"].(string)
		if !ok {
			return nil, fmt.Errorf("Unmarshal error. Key not a string: %v", data["key"])
		}

		// retrieve object
		model, err := d.Get(ds.NewKey(k))
		if err != nil {
			return nil, fmt.Errorf("%v (ElasticSearch inconsistent?)", err)
		}

		// ok! we have it.
		models = append(models, model)
	}
	return &models, nil
}

// for testing
func main() {
	flag.Parse()
	log.SetFlags(log.Ltime | log.Lshortfile)
	api.Domain = *eshost
	response, _ := core.Index(true, "twitter", "tweet", "1", NewTweet("kimchy", "Search is cool"))
	indices.Flush()
	log.Printf("Index OK: %v", response.Ok)
	searchresponse, err := core.SearchRequest(true, "twitter", "tweet", "{\"query\" : {\"term\" : { \"user\" : \"kimchy\" }}}", "", 0)
	if err != nil {
		log.Println("error during search:" + err.Error())
		log.Fatal(err)
	}
	// try marshalling to tweet type
	var t Tweet
	json.Unmarshal(searchresponse.Hits.Hits[0].Source, t)
	log.Printf("Search Found: %s", t)
	response, _ = core.Get(true, "twitter", "tweet", "1")
	log.Printf("Get: %v", response.Exists)
	exists, _ := core.Exists(true, "twitter", "tweet", "1")
	log.Printf("Exists: %v", exists)
	indices.Flush()
	countResponse, _ := core.Count(true, "twitter", "tweet")
	log.Printf("Count: %v", countResponse.Count)
	response, _ = core.Delete(true, "twitter", "tweet", "1", -1, "")
	log.Printf("Delete OK: %v", response.Ok)
	response, _ = core.Get(true, "twitter", "tweet", "1")
	log.Printf("Get: %v", response.Exists)

	healthResponse, _ := cluster.Health(true)
	log.Printf("Health: %v", healthResponse.Status)

	cluster.State("transient", "discovery.zen.minimum_master_nodes", 2)

}

func main() {
	core.DebugRequests = true
	log.SetFlags(log.LstdFlags)
	flag.Parse()

	fmt.Println("host = ", *host)
	// Set the Elasticsearch Host to Connect to
	api.Domain = *host

	// Index a document
	_, err := core.Index("testindex", "user", "docid_1", nil, `{"name":"bob"}`)
	exitIfErr(err)

	// Index a doc using a map of values
	_, err = core.Index("testindex", "user", "docid_2", nil, map[string]string{"name": "venkatesh"})
	exitIfErr(err)

	// Index a doc using Structs
	_, err = core.Index("testindex", "user", "docid_3", nil, MyUser{"wanda", 22})
	exitIfErr(err)

	// Search Using Raw json String
	searchJson := `{
	    "query" : {
	        "term" : { "Name" : "wanda" }
	    }
	}`
	out, err := core.SearchRequest("testindex", "user", nil, searchJson)
	if len(out.Hits.Hits) == 1 {
		fmt.Println("%v", out.Hits.Hits[0].Source)
	}
	exitIfErr(err)

}

func search(query string) []Word {
	fmt.Println("Searching for... ", query)
	api.Domain = "localhost"
	searchJson := fmt.Sprintf(`{"query": {"multi_match": {"query": "%s", "fields": ["japanese", "furigana", "romaji", "english"]}}, "highlight": {"fields": {"furigana": {}, "japanese": {}, "romaji": {}, "english": {}}}}`, query)
	out, err := core.SearchRequest(true, "edict", "entry", searchJson, "", 0)
	if err != nil {
		log.Println(err)
	}

	hits := [][]byte{}
	for _, hit := range out.Hits.Hits {
		hits = append(hits, hit.Source)
	}

	wordList := []Word{}
	for _, hit := range hits {
		w := Word{}
		err := json.Unmarshal(hit, &w)
		if err != nil {
			log.Println(err)
		}
		wordList = append(wordList, w)
	}
	return wordList
}

func GetFeels() {

	api.Domain = "localhost"

	// search users
	var searchQuery = `{
		"query": {
			"term": {"firstName":"J"}
		}
	}`

	response, err := core.SearchRequest("gccount", "Customer", nil, searchQuery)
	if err != nil {
		log.Fatalf("The search of users has failed:", err)
	}
	//	var values []interface{}

	for _, hit := range response.Hits.Hits {
		/*
			var value map[string]interface{}
			err := json.Unmarshal(hit.Source, &value)
			if err != nil {
				log.Fatalf("Failed to unmarshal, line 43", err)
			}
			values = append(values, value)
		*/
		fmt.Println(hit)
	}
	//fmt.Println(values)

}

func executeQuery(index, queryFile string) (core.SearchResult, error) {
	f, err := os.Open(filepath.Join(QueriesStore, queryFile))
	if err != nil {
		return core.SearchResult{}, fmt.Errorf("fail to open query file %q", queryFile)
	}
	defer f.Close()
	return core.SearchRequest(index, "", map[string]interface{}{}, f)
}

func TestSearchRequest(t *testing.T) {
	qry := map[string]interface{}{
		"query": map[string]interface{}{
			"wildcard": map[string]string{"actor": "a*"},
		},
	}
	out, err := core.SearchRequest(true, "github", "", qry, "", 0)
	//log.Println(out)
	assert.T(t, &out != nil && err == nil, t, "Should get docs")
	assert.T(t, out.Hits.Total == 616 && out.Hits.Len() == 10, t, "Should have 616 hits but was %v", out.Hits.Total)
}

func TestSearchRequest(t *testing.T) {
	qry := map[string]interface{}{
		"query": map[string]interface{}{
			"wildcard": map[string]string{"actor": "a*"},
		},
	}
	out, err := core.SearchRequest("github", "", nil, qry)
	//log.Println(out)
	assert.T(t, &out != nil && err == nil, t, "Should get docs")
	expectedDocs := 10
	expectedHits := 621
	assert.T(t, out.Hits.Len() == expectedDocs, t, fmt.Sprintf("Should have %v docs but was %v", expectedDocs, out.Hits.Len()))
	assert.T(t, out.Hits.Total == expectedHits, t, fmt.Sprintf("Should have %v hits but was %v", expectedHits, out.Hits.Total))
}

func main() {

	type Address struct {
		ProvinceName string `json:"province_name"`
	}

	api.Domain = "localhost"

	searchJson := `{"query":{"match":{"province_name":"ไทย"}}}`

	out, _ := core.SearchRequest("data", "address", nil, searchJson)
	for i := 0; i < len(out.Hits.Hits); i++ {
		fmt.Printf("%s\n", out.Hits.Hits[i].Source)
	}

}

func ReadEsMetric(name string) (our_el *chains.ChainEl, err error) {
	our_el = chains.NewChainEl()
	go func(our_el *chains.ChainEl) {
		from := <-our_el.Settings
		until := <-our_el.Settings
		qry := map[string]interface{}{
			"query": map[string]interface{}{
				"term": map[string]string{"metric": name},
				"range": map[string]interface{}{
					"ts": map[string]string{"from": strconv.Itoa(int(from)), "to": strconv.Itoa(int(until))},
				},
			},
		}
		// { "bool": { "must": [ {"term": ... }, {"range": ...}] }}

		// TODO: sorting?
		out, err := core.SearchRequest(true, "carbon-es", "datapoint", qry, "", 0)
		if err != nil {
			panic(fmt.Sprintf("error reading ES for %s: %s", name, err.Error()))

		}
		// if we don't have enough data to cover the requested timespan, fill with nils
		/* if metric.Data[0].Ts > from {
			for new_ts := from; new_ts < metric.Data[0].Ts; new_ts += 60 {
				our_el.Link <- *metrics.NewDatapoint(new_ts, 0.0, false)
			}
		}
		for _, d := range metric.Data {
			if d.Ts >= from && until <= until {
				our_el.Link <- *d
			}
		}
		if metric.Data[len(metric.Data)-1].Ts < until {
			for new_ts := metric.Data[len(metric.Data)-1].Ts + 60; new_ts <= until+60; new_ts += 60 {
				our_el.Link <- *metrics.NewDatapoint(new_ts, 0.0, false)
			}
		}
		*/

		fmt.Println(out)
	}(our_el)
	return our_el, nil
}

func Search(query, lang string) (sb []SearchResult, err error) {

	qry := `{
		"query": {
			"term": { "keywords": "` + query + `" }
		}
	}`

	out, err := core.SearchRequest(IndexName, lang, nil, qry)

	for _, hit := range out.Hits.Hits {

		var obj SearchResult
		json.Unmarshal(*hit.Source, &obj)

		sb = append(sb, obj)
	}

	return sb, err
}

// for testing
func main() {
	flag.Parse()
	log.SetFlags(log.Ltime | log.Lshortfile)
	api.Domain = *eshost
	core.VerboseLogging = true
	response, _ := core.Index("twitter", "tweet", "1", nil, NewTweet("kimchy", "Search is cool"))
	indices.Flush()
	log.Printf("Index OK: %v", response.Ok)
	searchresponse, err := core.SearchRequest("twitter", "tweet", nil, "{\"query\" : {\"term\" : { \"user\" : \"kimchy\" }}}")
	if err != nil {
		log.Println("error during search:" + err.Error())
		log.Fatal(err)
	}
	// try marshalling to tweet type
	var t Tweet
	bytes, err := searchresponse.Hits.Hits[0].Source.MarshalJSON()
	if err != nil {
		log.Fatalf("err calling marshalJson:%v", err)
	}
	json.Unmarshal(bytes, t)
	log.Printf("Search Found: %s", t)
	response, _ = core.Get("twitter", "tweet", "1", nil)
	log.Printf("Get: %v", response.Exists)
	exists, _ := core.Exists("twitter", "tweet", "1", nil)
	log.Printf("Exists: %v", exists)
	indices.Flush()
	countResponse, _ := core.Count("twitter", "tweet", nil)
	log.Printf("Count: %v", countResponse.Count)
	response, _ = core.Delete("twitter", "tweet", "1", map[string]interface{}{"version": -1, "routing": ""})
	log.Printf("Delete OK: %v", response.Ok)
	response, _ = core.Get("twitter", "tweet", "1", nil)
	log.Printf("Get: %v", response.Exists)

	healthResponse, _ := cluster.Health()
	log.Printf("Health: %v", healthResponse.Status)

	cluster.UpdateSettings("transient", "discovery.zen.minimum_master_nodes", 2)

}

func main() {
	api.Domain = os.Getenv("ES_ENDPOINT")
	api.Port = os.Getenv("ES_HTTP_PORT")

	index := "production_v4"

	m := martini.Classic()
	m.Use(render.Renderer(render.Options{
		Layout: "layout",
	}))

	m.Get("/", func(r render.Render) {

		out, _ := ecCore.SearchRequest(index, "info", nil, nil)

		if len(out.Hits.Hits) > 0 {

			infos := []core.SyncInfo{}

			sem := make(async.Semaphore, out.Hits.Len())

			for _, hit := range out.Hits.Hits {
				go func(hit ecCore.Hit) {

					var syncInfo core.SyncInfo
					json.Unmarshal(*hit.Source, &syncInfo)

					searchJson := fmt.Sprintf(`{
                        "sort" : [
                            { "reputation" : {"order" : "desc"}}
                        ],
                        query: {
                          match: {
                              "_ninya_sync_task_id": %v
                          }
                        }
                    }`, syncInfo.TaskId)

					taskInfo, _ := ecCore.SearchRequest(index, "user", nil, searchJson)
					syncInfo.SyncedEntities = taskInfo.Hits.Total
					syncInfo.Index = len(infos) + 1

					taskStartedAt := syncInfo.TaskId / 1000

					const layout = "Jan 2, 2006 at 3:04pm (MST)"
					syncInfo.TaskStarted = time.Unix(int64(taskStartedAt), 0).Format(layout)

					secondsElapsed := (int(time.Now().Unix()) - taskStartedAt)
					syncInfo.ElapsedTime = format.Duration(secondsElapsed)

					syncInfo.EntitiesPerHour = int(float32(syncInfo.SyncedEntities) / float32(secondsElapsed) * 60 * 60)
					infos = append(infos, syncInfo)

					sem.Signal()

				}(hit)
			}

			// what's going on here?! Our semaphone expects a the doubled total? Why?!
			sem.Wait(out.Hits.Total * 2)
			r.HTML(200, "syncList", infos)
		}
	})

	m.Run()
}

func GeoJotsHandler(w http.ResponseWriter, r *http.Request) {
	w.Header().Set("Content-Type", "application/json")

	err := r.ParseForm()
	if err != nil {
		serveError(w, err)
		return
	}

	lat, err := strconv.ParseFloat(r.Form.Get("lat"), 64)
	if err != nil {
		serveError(w, err)
		return
	}

	lon, err := strconv.ParseFloat(r.Form.Get("lon"), 64)
	if err != nil {
		serveError(w, err)
		return
	}

	// Find all GeoJots that are close by
	qry := map[string]interface{}{
		"from": 0,
		"size": 50,
		"query": map[string]interface{}{
			"filtered": map[string]interface{}{
				"query": map[string]interface{}{
					"match_all": map[string]interface{}{},
				},
				"filter": map[string]interface{}{
					"geo_distance": map[string]interface{}{
						"distance": "1km",
						"geo_jot.location": map[string]interface{}{
							"lat": lat,
							"lon": lon,
						},
					},
				},
			},
		},
	}

	resp, err := core.SearchRequest(true, es_index, es_index_type, qry, "")
	//resp, err := core.SearchUri(es_index, es_index_type, "name:*", "")

	if err != nil {
		serveError(w, err)
	}

	var jots []GeoJot
	if resp.Hits.Total > 0 {
		for _, value := range resp.Hits.Hits {
			var jot GeoJot
			if err := json.Unmarshal(value.Source, &jot); err != nil {
				serveError(w, err)
				return
			}
			jot.Id = value.Id
			jots = append(jots, jot)
		}
	}

	j, err := json.Marshal(GeoJotsJSON{GeoJots: jots})
	if err != nil {
		serveError(w, err)
	}
	w.Write(j)
}

func (publisher *PublisherType) PublishTopology(params ...string) error {

	var localAddrs []string = params

	if len(params) == 0 {
		addrs, err := LocalAddrs()
		if err != nil {
			ERR("Getting local IP addresses fails with: %s", err)
			return err
		}
		localAddrs = addrs
	}

	// delete old IP addresses
	searchJson := fmt.Sprintf("{query: {term: {name: %s}}}", strconv.Quote(publisher.name))
	res, err := core.SearchRequest("packetbeat-topology", "server-ip", nil, searchJson)
	if err == nil {
		for _, server := range res.Hits.Hits {

			var top Topology
			err = json.Unmarshal([]byte(*server.Source), &top)
			if err != nil {
				ERR("Failed to unmarshal json data: %s", err)
			}
			if !stringInSlice(top.Ip, localAddrs) {
				res, err := core.Delete("packetbeat-topology", "server-ip" /*id*/, top.Ip, nil)
				if err != nil {
					ERR("Failed to delete the old IP address from packetbeat-topology")
				}
				if !res.Ok {
					ERR("Fail to delete old topology entry")
				}
			}

		}
	}

	// add new IP addresses
	for _, addr := range localAddrs {

		// check if the IP is already in the elasticsearch, before adding it
		found, err := core.Exists("packetbeat-topology", "server-ip" /*id*/, addr, nil)
		if err != nil {
			ERR("core.Exists fails with: %s", err)
		} else {

			if !found {
				res, err := core.Index("packetbeat-topology", "server-ip" /*id*/, addr, nil,
					Topology{publisher.name, addr})
				if err != nil {
					return err
				}
				if !res.Ok {
					ERR("Fail to add new topology entry")
				}
			}
		}
	}

	DEBUG("publish", "Topology: name=%s, ips=%s", publisher.name, strings.Join(localAddrs, " "))

	// initialize local topology map
	publisher.TopologyMap = make(map[string]string)

	return nil
}

// Retrieves metrics that apply to the instance
func (self *Instance) GetAllMetrics() (*model.Metrics, error) {
	huddle := self.huddle

	jujuUnitName := self.jujuPrefix + "metrics"

	es := huddle.SystemServices["elasticsearch"]
	if es == nil {
		return nil, rs.ErrNotFound()
	}

	// TODO: Inject
	// TODO: Use an ES client that isn't a singleton
	elasticgo_api.Domain = es.PublicAddress
	elasticgo_api.Port = "9200"

	// TODO: We need to make sure that most fields are _not_ analyzed
	// That is why we have match below, not term
	query := map[string]interface{}{
		"query": map[string]interface{}{
			"match": map[string]string{"Hostname": jujuUnitName},
		},
	}

	//	query := map[string]interface{}{
	//		"query": map[string]interface{}{
	//			"match_all": map[string]string {},
	//		},
	//	}

	args := map[string]interface{}{}
	args["size"] = 1000

	response, err := elastigo_core.SearchRequest("_all", "message", args, query)
	if err != nil {
		log.Warn("Error searching elasticsearch", err)
		return nil, fmt.Errorf("Error searching elasticsearch")
	}

	metrics := &model.Metrics{}
	metrics.Metric = []string{}

	for _, v := range response.Hits.Hits {
		// TODO: Are we serializing and deserializing here??
		json, err := v.Source.MarshalJSON()
		if err != nil {
			log.Warn("Error reading JSON", err)
			return nil, fmt.Errorf("Error searching elasticsearch")
		}

		m := string(json)

		metrics.Metric = append(metrics.Metric, m)

		//		var value map[string]interface{}
		//		err := json.Unmarshal(v.Source, &value)
		//		if err != nil {
		//			log.Warn("Error unmarshalling response", err)
		//			return nil, fmt.Errorf("Error searching elasticsearch")
		//		}
		//		values = append(values, value)
	}
	//	fmt.Println(values)

	return metrics, nil
}

func (self *Instance) readMetrics(jujuUnitNames []string, metricId string) (*model.MetricDataset, error) {
	instance := self

	keyValue := metricId
	keyTimestamp := "Timestamp"
	keyHostname := "Hostname"
	keyType := "Type"

	huddle := instance.huddle

	es := huddle.SystemServices["elasticsearch"]
	if es == nil {
		return nil, rs.ErrNotFound()
	}

	if len(jujuUnitNames) == 0 {
		return nil, nil
	}

	// TODO: Inject
	// TODO: Use an ES client that isn't a singleton
	elasticgo_api.Domain = es.PublicAddress
	elasticgo_api.Port = "9200"

	// TODO: We need to make sure that most fields are _not_ analyzed
	// That is why we have match below, not term

	filters := []interface{}{}

	{
		match := map[string]string{}
		match[keyHostname] = jujuUnitNames[0]
		filter := map[string]interface{}{"query": map[string]interface{}{"match": match}}
		filters = append(filters, filter)
	}

	{
		// TODO: Hard-coded
		match := map[string]string{}
		match[keyType] = "LoadAverage"
		filter := map[string]interface{}{"query": map[string]interface{}{"match": match}}
		filters = append(filters, filter)
	}

	if len(filters) > 1 {
		and := map[string]interface{}{"and": filters}
		filters = []interface{}{and}
	}

	match_all := map[string]interface{}{"match_all": map[string]string{}}
	filtered := map[string]interface{}{"filter": filters[0], "query": match_all}

	query := map[string]interface{}{"filtered": filtered}

	body := map[string]interface{}{"query": query}

	args := map[string]interface{}{}
	args["size"] = 1000

	response, err := elastigo_core.SearchRequest("_all", "message", args, body)
	if err != nil {
		log.Warn("Error searching elasticsearch", err)
		return nil, fmt.Errorf("Error searching elasticsearch")
	}

	metrics := &model.MetricDataset{}
	metrics.Points = []model.MetricDatapoint{}

	for _, hit := range response.Hits.Hits {
		// TODO: Are we serializing and deserializing here??
		jsonBytes, err := hit.Source.MarshalJSON()
		if err != nil {
			log.Warn("Error reading JSON", err)
			return nil, fmt.Errorf("Error searching elasticsearch")
		}

		//log.Info("Found metric: %v", string(jsonBytes))

		var value map[string]interface{}
		err = json.Unmarshal(jsonBytes, &value)
		if err != nil {
			log.Warn("Error unmarshalling response", err)
			return nil, fmt.Errorf("Error searching elasticsearch")
		}

		// Post-filter the ES results...
		// TODO: See if we can persuade ES to filter it correctly
		hostname, found := value[keyHostname]
		if !found {
			log.Debug("No hostname in %v", string(jsonBytes))
			continue
		}

		hostnameStr, ok := hostname.(string)
		if !ok {
			log.Debug("Cannot cast hostname to string: %v", hostname)
			continue
		}

		if hostnameStr != jujuUnitNames[0] {
//.........这里部分代码省略.........