func (mb *Metricbeat) Config(b *beat.Beat) error {

	mb.MbConfig = &MetricbeatConfig{}
	err := cfgfile.Read(mb.MbConfig, "")
	if err != nil {
		fmt.Println(err)
		logp.Err("Error reading configuration file: %v", err)
		return err
	}

	mb.ModulesConfig = &RawModulesConfig{}
	err = cfgfile.Read(mb.ModulesConfig, "")
	if err != nil {
		fmt.Println(err)
		logp.Err("Error reading configuration file: %v", err)
		return err
	}

	mb.MetricsConfig = &RawMetricsConfig{}
	err = cfgfile.Read(mb.MetricsConfig, "")
	if err != nil {
		fmt.Println(err)
		logp.Err("Error reading configuration file: %v", err)
		return err
	}

	logp.Info("Setup base and raw configuration for Modules and Metrics")
	// Apply the base configuration to each module and metric
	for moduleName, module := range helper.Registry {
		// Check if config for module exist. Only configured modules are loaded
		if _, ok := mb.MbConfig.Metricbeat.Modules[moduleName]; !ok {
			continue
		}
		module.BaseConfig = mb.MbConfig.getModuleConfig(moduleName)
		module.RawConfig = mb.ModulesConfig.Metricbeat.Modules[moduleName]
		module.Enabled = true

		for metricSetName, metricSet := range module.MetricSets {
			// Check if config for metricset exist. Only configured metricset are loaded
			if _, ok := mb.MbConfig.getModuleConfig(moduleName).MetricSets[metricSetName]; !ok {
				continue
			}
			metricSet.BaseConfig = mb.MbConfig.getModuleConfig(moduleName).MetricSets[metricSetName]
			metricSet.RawConfig = mb.MetricsConfig.Metricbeat.Modules[moduleName].MetricSets[metricSetName]
			metricSet.Enabled = true
		}
	}

	return nil
}

func (fs *FileSizeBeat) Config(b *beat.Beat) error {
	err := cfgfile.Read(&fs.config, "")
	if err != nil {
		logp.Err("Error reading configuration file: %v", err)
		return err
	}

	if fs.config.Input.Period != nil {
		fs.period = time.Duration(*fs.config.Input.Period) * time.Second
	} else {
		fs.period = 10 * time.Second
	}
	logp.Debug("filesizebeat", "Period %v\n", fs.period)

	if fs.config.Input.Paths != nil {
		//fs.paths = make([]Path, len(*fs.config.Input.Paths))
		for _, path := range *fs.config.Input.Paths {
			err := fs.AddPath(path)
			if err != nil {
				logp.Critical("Error: %v", err)
				os.Exit(1)
			}
		}
		logp.Debug("filesizebeat", "Paths : %v\n", fs.paths)
	} else {
		logp.Critical("Error: no paths specified, cannot continue!")
		os.Exit(1)
	}
	return nil
}

func (bt *Govarbeat) Config(b *beat.Beat) error {

	cfg := &config.GovarbeatConfig{}
	err := cfgfile.Read(&cfg, "")
	if err != nil {
		return fmt.Errorf("Error reading config file: %v", err)
	}

	for name, cfg := range cfg.Remotes {
		defaultDuration := 1 * time.Second
		d := defaultDuration
		if cfg.Period != "" {
			d, err = time.ParseDuration(cfg.Period)
			if err != nil {
				return err
			}
		}

		bt.workers = append(bt.workers, &worker{
			done:   bt.done,
			host:   cfg.Host,
			name:   name,
			period: d,
		})
	}

	if len(bt.workers) == 0 {
		return fmt.Errorf("No workers configured")
	}

	return nil
}

func TestReadConfig(t *testing.T) {
	absPath, err := filepath.Abs("../tests/files/")

	assert.NotNil(t, absPath)
	assert.Nil(t, err)

	config := &ConfigSettings{}

	err = cfgfile.Read(config, absPath+"/config.yml")
	assert.Nil(t, err)

	execs := config.Execbeat.Execs
	assert.Equal(t, 2, len(execs))

	assert.Equal(t, "echo1", execs[0].Command)
	assert.Equal(t, "Execbeat", execs[0].Args)
	assert.Equal(t, "@every 1m", execs[0].Cron)
	assert.Equal(t, 2, len(execs[0].Fields))
	assert.Equal(t, "exec", execs[0].DocumentType)

	assert.Equal(t, "echo2", execs[1].Command)
	assert.Equal(t, "Hello World", execs[1].Args)
	assert.Equal(t, "@every 2m", execs[1].Cron)
	assert.Equal(t, 0, len(execs[1].Fields))
	assert.Equal(t, "", execs[1].DocumentType)
}

// Loads the beat specific config and overwrites params based on cmd line
func (pb *Packetbeat) Config(b *beat.Beat) error {

	// Read beat implementation config as needed for setup
	err := cfgfile.Read(&pb.PbConfig, "")

	// CLI flags over-riding config
	if *pb.CmdLineArgs.TopSpeed {
		pb.PbConfig.Interfaces.TopSpeed = true
	}

	if len(*pb.CmdLineArgs.File) > 0 {
		pb.PbConfig.Interfaces.File = *pb.CmdLineArgs.File
	}

	pb.PbConfig.Interfaces.Loop = *pb.CmdLineArgs.Loop
	pb.PbConfig.Interfaces.OneAtATime = *pb.CmdLineArgs.OneAtAtime

	if len(*pb.CmdLineArgs.Dumpfile) > 0 {
		pb.PbConfig.Interfaces.Dumpfile = *pb.CmdLineArgs.Dumpfile
	}

	// assign global singleton as it is used in protocols
	// TODO: Refactor
	config.ConfigSingleton = pb.PbConfig

	return err
}

func (bt *Generatorbeat) Config(b *beat.Beat) error {

	cfg := &config.GeneratorbeatConfig{}
	err := cfgfile.Read(&cfg, "")
	if err != nil {
		return fmt.Errorf("Error reading config file: %v", err)
	}

	for name, cfg := range cfg.Generators {
		factory, ok := generators[name]
		if !ok {
			return fmt.Errorf("Unknown generator: %v", name)
		}

		generators, err := factory(cfg)
		if err != nil {
			return err
		}

		for _, gen := range generators {
			bt.worker = append(bt.worker, &worker{
				done: bt.done,
				gen:  gen,
			})
		}
	}

	return nil
}

func TestReadConfig(t *testing.T) {
	absPath, err := filepath.Abs("../tests/files/")

	assert.NotNil(t, absPath)
	assert.Nil(t, err)

	config := &Config{}

	err = cfgfile.Read(config, absPath+"/config.yml")

	assert.Nil(t, err)

	assert.Equal(t, uint64(DefaultSpoolSize), config.Filebeat.SpoolSize)
	assert.Equal(t, "/prospectorConfigs/", config.Filebeat.ConfigDir)

	prospectors := config.Filebeat.Prospectors

	// Check if multiple paths were read in
	assert.Equal(t, 3, len(prospectors))

	// Check if full array can be read. Assumed that are ordered same as in config file
	assert.Equal(t, 2, len(prospectors[0].Paths))
	assert.Equal(t, "/var/log/s*.log", prospectors[0].Paths[1])
	assert.Equal(t, "log", prospectors[0].Input)
	assert.Equal(t, 3, len(prospectors[0].Harvester.Fields))
	assert.Equal(t, int64(1), prospectors[0].Harvester.Fields["review"])
	assert.Equal(t, "0", prospectors[0].IgnoreOlder)
	assert.Equal(t, "1h", prospectors[0].Harvester.CloseOlder)
	assert.Equal(t, "10s", prospectors[0].ScanFrequency)

	assert.Equal(t, "stdin", prospectors[2].Input)
	assert.Equal(t, 0, len(prospectors[2].Paths))
	assert.Equal(t, "", prospectors[1].ScanFrequency)
}

// LoadConfig inits the config file and reads the default config information
// into Beat.Config. It exists the processes in case of errors.
func (b *Beat) LoadConfig() error {

	err := cfgfile.Read(&b.Config, "")
	if err != nil {
		return fmt.Errorf("loading config file error: %v\n", err)
	}

	err = logp.Init(b.Name, &b.Config.Logging)
	if err != nil {
		return fmt.Errorf("error initializing logging: %v\n", err)
	}

	// Disable stderr logging if requested by cmdline flag
	logp.SetStderr()

	logp.Debug("beat", "Initializing output plugins")

	pub, err := publisher.New(b.Name, b.Config.Output, b.Config.Shipper)
	if err != nil {
		return fmt.Errorf("error Initialising publisher: %v\n", err)
	}

	b.Events = pub.Client()

	logp.Info("Init Beat: %s; Version: %s", b.Name, b.Version)

	return nil
}

func (eb *Winlogbeat) Config(b *beat.Beat) error {
	// Read configuration.
	err := cfgfile.Read(&eb.config, "")
	if err != nil {
		return fmt.Errorf("Error reading configuration file. %v", err)
	}

	// Validate configuration.
	err = eb.config.Winlogbeat.Validate()
	if err != nil {
		return fmt.Errorf("Error validating configuration file. %v", err)
	}
	debugf("Configuration validated. config=%v", eb.config)

	// Registry file grooming.
	if eb.config.Winlogbeat.RegistryFile == "" {
		eb.config.Winlogbeat.RegistryFile = config.DefaultRegistryFile
	}
	eb.config.Winlogbeat.RegistryFile, err = filepath.Abs(
		eb.config.Winlogbeat.RegistryFile)
	if err != nil {
		return fmt.Errorf("Error getting absolute path of registry file %s. %v",
			eb.config.Winlogbeat.RegistryFile, err)
	}
	logp.Info("State will be read from and persisted to %s",
		eb.config.Winlogbeat.RegistryFile)

	return nil
}

func (d *Dockerbeat) Config(b *beat.Beat) error {

	// Requires Docker 1.5 minimum
	d.minimalDockerVersion = SoftwareVersion{major: 1, minor: 5}

	err := cfgfile.Read(&d.TbConfig, "")
	if err != nil {
		logp.Err("Error reading configuration file: %v", err)
		return err
	}

	//init the period
	if d.TbConfig.Input.Period != nil {
		d.period = time.Duration(*d.TbConfig.Input.Period) * time.Second
	} else {
		d.period = 1 * time.Second
	}
	//init the socket
	if d.TbConfig.Input.Socket != nil {
		d.socket = *d.TbConfig.Input.Socket
	} else {
		d.socket = "unix:///var/run/docker.sock" // default docker socket location
	}

	logp.Info("dockerbeat", "Init dockerbeat")
	logp.Info("dockerbeat", "Follow docker socket %q\n", d.socket)
	logp.Info("dockerbeat", "Period %v\n", d.period)

	return nil
}

func (d *Dockerbeat) Config(b *beat.Beat) error {

	err := cfgfile.Read(&d.TbConfig, "")
	if err != nil {
		logp.Err("Error reading configuration file: %v", err)
		return err
	}

	//init the period
	if d.TbConfig.Input.Period != nil {
		d.period = time.Duration(*d.TbConfig.Input.Period) * time.Second
	} else {
		d.period = 1 * time.Second
	}
	//init the socket
	if d.TbConfig.Input.Socket != nil {
		d.socket = *d.TbConfig.Input.Socket
	} else {
		d.socket = "unix:///var/run/docker.sock" // default docker socket location
	}

	logp.Debug("dockerbeat", "Init dockerbeat")
	logp.Debug("dockerbeat", "Follow docker socket %q\n", d.socket)
	logp.Debug("dockerbeat", "Period %v\n", d.period)

	return nil
}

// LoadConfig inits the config file and reads the default config information
// into Beat.Config. It exists the processes in case of errors.
func (b *Beat) LoadConfig() {

	err := cfgfile.Read(&b.Config, "")
	if err != nil {
		// logging not yet initialized, so using fmt.Printf
		fmt.Printf("Loading config file error: %v\n", err)
		os.Exit(1)
	}

	err = logp.Init(b.Name, &b.Config.Logging)
	if err != nil {
		fmt.Printf("Error initializing logging: %v\n", err)
		os.Exit(1)
	}

	// Disable stderr logging if requested by cmdline flag
	logp.SetStderr()

	logp.Debug("beat", "Initializing output plugins")

	pub, err := publisher.New(b.Name, b.Config.Output, b.Config.Shipper)
	if err != nil {
		fmt.Printf("Error Initialising publisher: %v\n", err)
		logp.Critical(err.Error())
		os.Exit(1)
	}

	b.Events = pub.Client()

	logp.Info("Init Beat: %s; Version: %s", b.Name, b.Version)
}

func (ub *Unifiedbeat) Config(b *beat.Beat) error {
	// load the unifiedbeat.yml config file
	err := cfgfile.Read(&ub.UbConfig, "")
	if err != nil {
		return fmt.Errorf("Error reading config file: %v", err)
	}
	return nil
}

func (bt *Kafkabeat) Config(b *beat.Beat) error {

	// Load beater beatConfig
	logp.Info("Configuring Kafkabeat...")
	var err error
	err = cfgfile.Read(&bt.beatConfig, "")
	if err != nil {
		return fmt.Errorf("Error reading config file: %v", err)
	}

	bt.zookeepers = bt.beatConfig.Kafkabeat.Zookeepers
	if bt.zookeepers == nil || len(bt.zookeepers) == 0 {
		return KafkabeatError{"Atleast one zookeeper must be defined"}
	}
	zClient, err = kazoo.NewKazoo(bt.zookeepers, nil)
	if err != nil {
		logp.Err("Unable to connect to Zookeeper")
		return err
	}
	bt.brokers, err = zClient.BrokerList()
	if err != nil {
		logp.Err("Error identifying brokers from zookeeper")
		return err
	}
	if bt.brokers == nil || len(bt.brokers) == 0 {
		return KafkabeatError{"Unable to identify active brokers"}
	}
	logp.Info("Brokers: %v", bt.brokers)
	client, err = sarama.NewClient(bt.brokers, sarama.NewConfig())

	// Setting default period if not set
	if bt.beatConfig.Kafkabeat.Period == "" {
		bt.beatConfig.Kafkabeat.Period = "1s"
	}

	bt.period, err = time.ParseDuration(bt.beatConfig.Kafkabeat.Period)
	if err != nil {
		return err
	}
	bt.topics = bt.beatConfig.Kafkabeat.Topics
	bt.create_topic_docs = true
	if bt.topics == nil || len(bt.topics) == 0 {
		bt.create_topic_docs = bt.topics == nil
		bt.topics, err = client.Topics()
	}
	if err != nil {
		return err
	}
	logp.Info("Monitoring topics: %v", bt.topics)
	bt.groups = bt.beatConfig.Kafkabeat.Groups
	if bt.groups == nil {
		bt.groups, err = getGroups()
	}
	logp.Info("Monitoring groups %v", bt.groups)
	return err
}

func (bt *Govarbeat) Config(b *beat.Beat) error {

	// Load beater configuration
	err := cfgfile.Read(&bt.Configuration, "")
	if err != nil {
		return fmt.Errorf("Error reading config file: %v", err)
	}

	return nil
}

func (tb *Topbeat) Config(b *beat.Beat) error {

	err := cfgfile.Read(&tb.TbConfig, "")
	if err != nil {
		logp.Err("Error reading configuration file: %v", err)
		return err
	}

	if tb.TbConfig.Input.Period != nil {
		tb.period = time.Duration(*tb.TbConfig.Input.Period) * time.Second
	} else {
		tb.period = 10 * time.Second
	}
	if tb.TbConfig.Input.Procs != nil {
		tb.procs = *tb.TbConfig.Input.Procs
	} else {
		tb.procs = []string{".*"} //all processes
	}

	if tb.TbConfig.Input.Stats.System != nil {
		tb.sysStats = *tb.TbConfig.Input.Stats.System
	} else {
		tb.sysStats = true
	}
	if tb.TbConfig.Input.Stats.Proc != nil {
		tb.procStats = *tb.TbConfig.Input.Stats.Proc
	} else {
		tb.procStats = true
	}
	if tb.TbConfig.Input.Stats.Filesystem != nil {
		tb.fsStats = *tb.TbConfig.Input.Stats.Filesystem
	} else {
		tb.fsStats = true
	}
	if tb.TbConfig.Input.Stats.CpuPerCore != nil {
		tb.cpuPerCore = *tb.TbConfig.Input.Stats.CpuPerCore
	} else {
		tb.cpuPerCore = false
	}

	if !tb.sysStats && !tb.procStats && !tb.fsStats {
		return errors.New("Invalid statistics configuration")
	}

	logp.Debug("topbeat", "Init topbeat")
	logp.Debug("topbeat", "Follow processes %q\n", tb.procs)
	logp.Debug("topbeat", "Period %v\n", tb.period)
	logp.Debug("topbeat", "System statistics %t\n", tb.sysStats)
	logp.Debug("topbeat", "Process statistics %t\n", tb.procStats)
	logp.Debug("topbeat", "File system statistics %t\n", tb.fsStats)
	logp.Debug("topbeat", "Cpu usage per core %t\n", tb.cpuPerCore)

	return nil
}

func (h *Httpbeat) Config(b *beat.Beat) error {

	err := cfgfile.Read(&h.HbConfig, "")
	if err != nil {
		logp.Err("Error reading configuration file: %v", err)
		return err
	}

	logp.Info("httpbeat", "Init httpbeat")

	return nil
}

func (execBeat *Execbeat) Config(b *beat.Beat) error {

	err := cfgfile.Read(&execBeat.ExecConfig, "")
	if err != nil {
		logp.Err("Error reading configuration file: %v", err)
		return err
	}

	logp.Info("execbeat", "Init execbeat")

	return nil
}

func (nagiosCheckBeat *NagiosCheckBeat) Config(b *beat.Beat) error {
	var config config.ConfigSettings
	err := cfgfile.Read(&config, "")
	if err != nil {
		logp.Err("Error reading configuration file: %v", err)
		return err
	}

	nagiosCheckBeat.checks = config.Input.Checks

	return nil
}

func New(b *beat.Beat, cfg *common.Config) (beat.Beater, error) {
	bt := &Httpbeat{
		done: make(chan bool),
	}

	err := cfgfile.Read(&bt.HbConfig, "")
	if err != nil {
		logp.Err("Error reading config file: %v", err)
		return nil, fmt.Errorf("Error reading config file: %v", err)
	}

	return bt, nil
}