func (im *IndexManager) processMetricPath(splitPath []string, pathLen int, isLeaf bool) {
	retries := 0
	for pathLen > 0 {
		metricPath := strings.Join(splitPath, ".")
		if string(metricPath[len(metricPath)-1]) == "%" {
			metricPath = metricPath[:len(metricPath)-1]
		}
		config.G.Log.System.LogDebug("IndexManager indexing \"%s\"", metricPath)
		pathToIndex := IndexResponse{
			metricPath,
			pathLen,
			"",
			isLeaf,
		}
		urlToPut := strings.Join([]string{config.G.ElasticSearch.PutURL, metricPath}, "/")
		jsonPath, err := json.Marshal(pathToIndex)
		if err != nil {
			logging.Statsd.Client.Inc("indexmgr.es.err.json", 1, 1.0)
			config.G.Log.System.LogError("Unable to marshal pathToIndex of %v, error is %v", pathToIndex, err.Error())
			return  
		}
		putreq, err := http.NewRequest("PUT", urlToPut, bytes.NewBuffer(jsonPath))
		if err != nil {
			logging.Statsd.Client.Inc("indexmgr.es.err.put", 1, 1.0)
			config.G.Log.System.LogError("Error when attempting to index %v: %v", metricPath, err.Error())
		}
		r := im.httpRequest(putreq)
		if r != nil {
			_, splitPath = splitPath[len(splitPath)-1], splitPath[:len(splitPath)-1]
			isLeaf = false
			pathLen = len(splitPath)
			retries = 0
		} else {
			logging.Statsd.Client.Inc("indexmgr.es.err.pmr.req", 1, 1.0)
			retries++
			config.G.Log.System.LogWarn("processMetricPath's httprequest to ES came back as nil, sending to retry in %d seconds.", retries)
			time.Sleep(time.Duration(retries) * time.Second)
		}
	}
}
processMetricPath recursively indexes the metric path via the ElasticSearch REST API.
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.")
	}
}
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
}
func (im *IndexManager) prepRequest(fullQuery ERQuery) *http.Request {
	jsonQuery, _ := json.Marshal(fullQuery)
	config.G.Log.System.LogDebug("%s", string(jsonQuery))
	countreq, _ := http.NewRequest("GET", config.G.ElasticSearch.CountURL, strings.NewReader(string(jsonQuery)))
	size := "size=" + im.getCount(countreq)
	searchURL := strings.Join([]string{config.G.ElasticSearch.SearchURL, size}, "?")
	getreq, _ := http.NewRequest("GET", searchURL, strings.NewReader(string(jsonQuery)))
	return getreq
}
func (im *IndexManager) index(path string) {
	it := time.Now()
	config.G.Log.System.LogDebug("IndexManager::index path=%s", path)
	splitPath := strings.Split(path, ".")
	im.processMetricPath(splitPath, len(splitPath), true)
	logging.Statsd.Client.TimingDuration("indexmgr.index", time.Since(it), 1.0)
}
func (im *IndexManager) httpRequest(req *http.Request) []byte {
	client := &http.Client{Timeout: time.Duration(15 * time.Second)}
	resp, err := client.Do(req)
	if err != nil {
		logging.Statsd.Client.Inc("indexmgr.es.err.httpreq", 1, 1.0)
		config.G.Log.System.LogError("Received error from ElasticSearch: %v, request: %v", err.Error(), req)
		return nil
	}
	defer resp.Body.Close()
	body, _ := ioutil.ReadAll(resp.Body)
	return body
}