func (im *IndexManager) getAllLeafNodes() []string {
	sort := []map[string]map[string]string{
		{
			"path": map[string]string{
				"order": "asc",
			},
		},
	}
	query := map[string]map[string][]map[string]map[string]interface{}{
		"bool": map[string][]map[string]map[string]interface{}{
			"must": []map[string]map[string]interface{}{
				{
					"match": map[string]interface{}{
						"leaf": true,
					},
				},
			},
		},
	}
	fullQuery := ERQuery{sort, query}
	getreq := im.prepRequest(fullQuery)
	r := im.httpRequest(getreq)
	var esResp ElasticResponse
	var pathList []string
	if r != nil {
		_ = json.Unmarshal(r, &esResp)
		config.G.Log.System.LogDebug("esResp: %v", esResp)
		for _, hit := range esResp.Hits.Hits {
			pathList = append(pathList, hit.Source.Path)
		}
	} else {
		logging.Statsd.Client.Inc("indexmgr.es.err.get", 1, 1.0)
		config.G.Log.System.LogError("Error querying ES.")
	}
	config.G.Log.System.LogDebug("Retrieved %v stat paths.", len(pathList))
	return pathList
}
getAllLeafNodes queries ElasticSearch for all leaf nodes. Used for populating metric manager's stat paths on reboot.
package datastore
import (
	"bytes"
	"encoding/json"
	"io/ioutil"
	"net/http"
	"strconv"
	"strings"
	"sync"
	"time"
	"github.com/otium/queue"
	"github.com/jeffpierce/cassabon/config"
	"github.com/jeffpierce/cassabon/logging"
)
type IndexResponse struct {
	Path   string `json:"path"`
	Depth  int    `json:"depth"`
	Tenant string `json:"tenant"`
	Leaf   bool   `json:"leaf"`
}
type ElasticResponse struct {
	Took     int      `json:"took"`
	TimedOut bool     `json:"timed_out"`
	Shards   ERShards `json:"_shards"`
	Hits     ERHits   `json:"hits"`
}
type ERShards struct {
	Total      int `json:"total"`
	Successful int `json:"successful"`
	Failed     int `json:"failed"`
}
type ERHits struct {
	Total    int           `json:"total"`
	MaxScore float32       `json:"max_score"`
	Hits     []ERSearchHit `json:"hits"`
}
type ERSearchHit struct {
	Index  string        `json:"_index"`
	Type   string        `json:"_type"`
	ID     string        `json:"_id"`
	Score  float32       `json:"_score"`
	Source IndexResponse `json:"_source"`
}
type ERQuery struct {
	Sort  []map[string]map[string]string                            `json:"sort"`
	Query map[string]map[string][]map[string]map[string]interface{} `json:"query"`
}
type IndexManager struct {
	wg         *sync.WaitGroup
	IndexQueue *queue.Queue
}
func (im *IndexManager) Init(bootstrap bool) {
	if bootstrap {
		im.initMapping()
	}
	im.IndexQueue = queue.NewQueue(func(metricPath interface{}) {
		if path, ok := metricPath.(string); ok {
			im.index(path)
		}
	}, 100)
}
func (im *IndexManager) Start(wg *sync.WaitGroup) {
	im.wg = wg
	im.wg.Add(1)
	go im.run()
}
func (im *IndexManager) run() {
	defer config.G.OnPanic()
	for {
		select {
		case <-config.G.OnReload2:
			config.G.Log.System.LogDebug("IndexManager::run received QUIT message")
			im.wg.Done()
			return
		case metric := <-config.G.Channels.IndexStore:
			im.IndexQueue.Push(metric.Path)
		case query := <-config.G.Channels.IndexRequest:
			go im.query(query)
		}
	}
}
func (im *IndexManager) initMapping() {
	mapping := map[string]map[string]map[string]map[string]map[string]string{
		"mappings": map[string]map[string]map[string]map[string]string{
			"path": map[string]map[string]map[string]string{
				"properties": map[string]map[string]string{
					"path": map[string]string{
						"type":  "string",
						"index": "not_analyzed",
					},
					"depth": map[string]string{
						"type": "long",
					},
					"tenant": map[string]string{
						"type": "string",
					},
					"leaf": map[string]string{
						"type": "boolean",
					},
				},
			},
		},
	}
	jsonMap, _ := json.Marshal(mapping)
	config.G.Log.System.LogDebug("%s", string(jsonMap))
	putreq, _ := http.NewRequest("PUT", config.G.ElasticSearch.MapURL, bytes.NewBuffer(jsonMap))
	r := im.httpRequest(putreq)
	config.G.Log.System.LogDebug("%v", string(r))
	if r == nil {
		config.G.Log.System.LogFatal("Could not initialize mapping for ElasticSearch.")
	}
}