func (out *elasticsearchOutput) loadTemplate(config Template, client *Client) error {
	out.templateMutex.Lock()
	defer out.templateMutex.Unlock()
	logp.Info("Trying to load template for client: %s", client.Connection.URL)
	exists := client.CheckTemplate(config.Name)
	if !exists || config.Overwrite {
		if config.Overwrite {
			logp.Info("Existing template will be overwritten, as overwrite is enabled.")
		}
		template := out.template
		if config.Versions.Es2x.Enabled && strings.HasPrefix(client.Connection.version, "2.") {
			logp.Info("Detected Elasticsearch 2.x. Automatically selecting the 2.x version of the template")
			template = out.template2x
		}
		err := client.LoadTemplate(config.Name, template)
		if err != nil {
			return fmt.Errorf("Could not load template: %v", err)
		}
	} else {
		logp.Info("Template already exists and will not be overwritten.")
	}
	return nil
}
loadTemplate checks if the index mapping template should be loaded
In case the template is not already loaded or overwritting is enabled, the
template is written to index
package elasticsearch
import (
	"encoding/json"
	"errors"
	"fmt"
	"net/url"
	"os"
	"strings"
	"sync"
	"time"
	"github.com/elastic/beats/libbeat/common"
	"github.com/elastic/beats/libbeat/common/op"
	"github.com/elastic/beats/libbeat/logp"
	"github.com/elastic/beats/libbeat/outputs"
	"github.com/elastic/beats/libbeat/outputs/mode"
	"github.com/elastic/beats/libbeat/outputs/mode/modeutil"
	"github.com/elastic/beats/libbeat/outputs/outil"
	"github.com/elastic/beats/libbeat/outputs/transport"
	"github.com/elastic/beats/libbeat/paths"
)
type elasticsearchOutput struct {
	index    outil.Selector
	beatName string
	pipeline *outil.Selector
	mode mode.ConnectionMode
	topology
	template      map[string]interface{}
	template2x    map[string]interface{}
	templateMutex sync.Mutex
}
func init() {
	outputs.RegisterOutputPlugin("elasticsearch", New)
}
var (
	debugf = logp.MakeDebug("elasticsearch")
)
var (
	ErrNotConnected = errors.New("not connected")
	ErrJSONEncodeFailed = errors.New("json encode failed")
	ErrResponseRead = errors.New("bulk item status parse failed")
)
func New(beatName string, cfg *common.Config, topologyExpire int) (outputs.Outputer, error) {
	if !cfg.HasField("bulk_max_size") {
		cfg.SetInt("bulk_max_size", -1, defaultBulkSize)
	}
	if !cfg.HasField("index") {
		pattern := fmt.Sprintf("%v-%%{+yyyy.MM.dd}", beatName)
		cfg.SetString("index", -1, pattern)
	}
	output := &elasticsearchOutput{beatName: beatName}
	err := output.init(cfg, topologyExpire)
	if err != nil {
		return nil, err
	}
	return output, nil
}
func (out *elasticsearchOutput) init(
	cfg *common.Config,
	topologyExpire int,
) error {
	config := defaultConfig
	if err := cfg.Unpack(&config); err != nil {
		return err
	}
	index, err := outil.BuildSelectorFromConfig(cfg, outil.Settings{
		Key:              "index",
		MultiKey:         "indices",
		EnableSingleOnly: true,
		FailEmpty:        true,
	})
	if err != nil {
		return err
	}
	tlsConfig, err := outputs.LoadTLSConfig(config.TLS)
	if err != nil {
		return err
	}
	err = out.readTemplate(&config.Template)
	if err != nil {
		return err
	}
	out.index = index
	pipeline, err := outil.BuildSelectorFromConfig(cfg, outil.Settings{
		Key:              "pipeline",
		MultiKey:         "pipelines",
		EnableSingleOnly: true,
		FailEmpty:        false,
	})
	if err != nil {
		return err
	}
	if !pipeline.IsEmpty() {
		out.pipeline = &pipeline
	}
	clients, err := modeutil.MakeClients(cfg, makeClientFactory(tlsConfig, &config, out))
	if err != nil {
		return err
	}
	maxRetries := config.MaxRetries
	maxAttempts := maxRetries + 1  
	if maxRetries < 0 {
		maxAttempts = 0
	}
	var waitRetry = time.Duration(1) * time.Second
	var maxWaitRetry = time.Duration(60) * time.Second
	out.clients = clients
	loadBalance := config.LoadBalance
	m, err := modeutil.NewConnectionMode(clients, modeutil.Settings{
		Failover:     !loadBalance,
		MaxAttempts:  maxAttempts,
		Timeout:      config.Timeout,
		WaitRetry:    waitRetry,
		MaxWaitRetry: maxWaitRetry,
	})
	if err != nil {
		return err
	}
	out.mode = m
	return nil
}
func (out *elasticsearchOutput) readTemplate(config *Template) error {
	if config.Enabled {
		if config.Name == "" {
			config.Name = out.beatName
		}
		if config.Path == "" {
			config.Path = fmt.Sprintf("%s.template.json", out.beatName)
		}
		if config.Versions.Es2x.Path == "" {
			config.Versions.Es2x.Path = fmt.Sprintf("%s.template-es2x.json", out.beatName)
		}
		templatePath := paths.Resolve(paths.Config, config.Path)
		logp.Info("Loading template enabled. Reading template file: %v", templatePath)
		template, err := readTemplate(templatePath)
		if err != nil {
			return fmt.Errorf("Error loading template %s: %v", templatePath, err)
		}
		out.template = template
		if config.Versions.Es2x.Enabled {
			templatePath := paths.Resolve(paths.Config, config.Versions.Es2x.Path)
			logp.Info("Loading template enabled for Elasticsearch 2.x. Reading template file: %v", templatePath)
			template, err := readTemplate(templatePath)
			if err != nil {
				return fmt.Errorf("Error loading template %s: %v", templatePath, err)
			}
			out.template2x = template
		}
	}
	return nil
}
func readTemplate(filename string) (map[string]interface{}, error) {
	f, err := os.Open(filename)
	if err != nil {
		return nil, err
	}
	defer f.Close()
	var template map[string]interface{}
	dec := json.NewDecoder(f)
	err = dec.Decode(&template)
	if err != nil {
		return nil, err
	}
	return template, nil
}