// PublishEvent forwards a single event. On failure PublishEvent tries to reconnect.
func (f *FailOverConnectionMode) PublishEvent(
	signaler outputs.Signaler,
	event common.MapStr,
) error {
	fails := 0
	var err error
	for !f.closed && (f.maxAttempts == 0 || fails < f.maxAttempts) {
		if err := f.connect(f.active); err != nil {
			logp.Info("Connecting error publishing events (retrying): %s", err)
			fails++
			time.Sleep(f.waitRetry)
			continue
		}

		if err := f.conns[f.active].PublishEvent(event); err != nil {
			logp.Info("Error publishing events (retrying): %s", err)
			fails++
			continue
		}

		outputs.SignalCompleted(signaler)
		return nil
	}

	outputs.SignalFailed(signaler, err)
	return nil
}

func (restart *ProspectorResume) Scan(files []cfg.FileConfig, persist map[string]*FileState, eventChan chan *FileEvent) {
	pendingProspectorCnt := 0

	// Prospect the globs/paths given on the command line and launch harvesters
	for _, fileconfig := range files {

		prospector := &Prospector{FileConfig: fileconfig}
		go prospector.Prospect(restart, eventChan)
		pendingProspectorCnt++
	}

	// Now determine which states we need to persist by pulling the events from the prospectors
	// When we hit a nil source a prospector had finished so we decrease the expected events
	logp.Info("filebeat", "Waiting for %d prospectors to initialise", pendingProspectorCnt)

	for event := range restart.Persist {
		if event.Source == nil {
			pendingProspectorCnt--
			if pendingProspectorCnt == 0 {
				break
			}
			continue
		}
		persist[*event.Source] = event
		logp.Info("filebeat", "Registrar will re-save state for %s", *event.Source)
	}

	logp.Info("filebeat", "All prospectors initialised with %d states to persist", len(persist))

}

func (r *Registrar) Run() {
	logp.Info("Starting Registrar")

	r.running = true

	// Writes registry on shutdown
	defer r.writeRegistry()

	for {
		select {
		case <-r.done:
			logp.Info("Ending Registrar")
			return
		// Treats new log files to persist with higher priority then new events
		case state := <-r.Persist:
			r.State[*state.Source] = state
			logp.Debug("prospector", "Registrar will re-save state for %s", *state.Source)
		case events := <-r.Channel:
			r.processEvents(events)
		}

		if e := r.writeRegistry(); e != nil {
			// REVU: but we should panic, or something, right?
			logp.Err("Writing of registry returned error: %v. Continuing..", e)
		}
	}
}

func (r *Registrar) fetchState(filePath string, fileInfo os.FileInfo) (int64, bool) {

	// Check if there is a state for this file
	lastState, isFound := r.GetFileState(filePath)

	if isFound && input.IsSameFile(filePath, fileInfo) {
		logp.Debug("registar", "Same file as before found. Fetch the state and persist it.")
		// We're resuming - throw the last state back downstream so we resave it
		// And return the offset - also force harvest in case the file is old and we're about to skip it
		r.Persist <- lastState
		return lastState.Offset, true
	}

	if previous, err := r.getPreviousFile(filePath, fileInfo); err == nil {
		// File has rotated between shutdown and startup
		// We return last state downstream, with a modified event source with the new file name
		// And return the offset - also force harvest in case the file is old and we're about to skip it
		logp.Info("Detected rename of a previously harvested file: %s -> %s", previous, filePath)

		lastState, _ := r.GetFileState(previous)
		lastState.Source = &filePath
		r.Persist <- lastState
		return lastState.Offset, true
	}

	if isFound {
		logp.Info("Not resuming rotated file: %s", filePath)
	}

	// New file so just start from an automatic position
	return 0, false
}

// PublishEvent forwards a single event. On failure PublishEvent tries to reconnect.
func (s *SingleConnectionMode) PublishEvent(
	signaler outputs.Signaler,
	event common.MapStr,
) error {
	fails := 0

	var err error

	for !s.closed && (s.maxAttempts == 0 || fails < s.maxAttempts) {
		if err = s.connect(); err != nil {
			logp.Info("Connecting error publishing event (retrying): %s", err)

			fails++
			time.Sleep(s.waitRetry)
			continue
		}
		if err := s.conn.PublishEvent(event); err != nil {
			logp.Info("Error publishing event (retrying): %s", err)

			fails++
			continue
		}

		outputs.SignalCompleted(signaler)
		return nil
	}

	outputs.SignalFailed(signaler, err)
	return nil
}

func (d *Dockerbeat) exportContainerStats(container docker.APIContainers) error {
	statsC := make(chan *docker.Stats)
	done := make(chan bool)
	errC := make(chan error, 1)
	statsOptions := docker.StatsOptions{container.ID, statsC, false, done, -1}
	go func() {
		logp.Info("1")
		errC <- d.dockerClient.Stats(statsOptions)
		logp.Info("2")
		close(errC)
	}()

	go func() {
		stats := <-statsC

		events := []common.MapStr{
			d.getContainerEvent(&container, stats),
			d.getCpuEvent(&container, stats),
			d.getMemoryEvent(&container, stats),
			d.getNetworkEvent(&container, stats),
		}

		d.events.PublishEvents(events)
	}()

	return nil
}

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

	d.isAlive = true

	var err error

	for d.isAlive {

		containers, err := d.dockerClient.ListContainers(docker.ListContainersOptions{})

		if err == nil {
			for _, container := range containers {
				logp.Info("3")
				d.exportContainerStats(container)
				logp.Info("4")
			}
			logp.Info("running! %s", d.socket)
		} else {
			logp.Err("Cannot get container list: %d", err)
		}
		time.Sleep(d.period)
	}

	return err
}

func (p *Prospector) calculateResume(file string, fileinfo os.FileInfo, resume *ProspectorResume) (int64, bool) {
	last_state, is_found := resume.Files[file]

	if is_found && IsSameFile(file, fileinfo, last_state) {
		// We're resuming - throw the last state back downstream so we resave it
		// And return the offset - also force harvest in case the file is old and we're about to skip it
		resume.Persist <- last_state
		return last_state.Offset, true
	}

	if previous := p.isFileRenamedResumelist(file, fileinfo, resume.Files); previous != "" {
		// File has rotated between shutdown and startup
		// We return last state downstream, with a modified event source with the new file name
		// And return the offset - also force harvest in case the file is old and we're about to skip it
		logp.Info("prospector", "Detected rename of a previously harvested file: %s -> %s", previous, file)
		last_state := resume.Files[previous]
		last_state.Source = &file
		resume.Persist <- last_state
		return last_state.Offset, true
	}

	if is_found {
		logp.Info("prospector", "Not resuming rotated file: %s", file)
	}

	// New file so just start from an automatic position
	return 0, false
}

// bulkCollectPublishFails checks per item errors returning all events
// to be tried again due to error code returned for that items. If indexing an
// event failed due to some error in the event itself (e.g. does not respect mapping),
// the event will be dropped.
func bulkCollectPublishFails(
	res *BulkResult,
	events []common.MapStr,
) []common.MapStr {
	failed := events[:0]
	for i, rawItem := range res.Items {
		status, msg, err := itemStatus(rawItem)
		if err != nil {
			logp.Info("Failed to parse bulk reponse for item (%i): %v", i, err)
			// add index if response parse error as we can not determine success/fail
			failed = append(failed, events[i])
			continue
		}

		if status < 300 {
			continue // ok value
		}

		if status < 500 && status != 429 {
			// hard failure, don't collect
			logp.Warn("Can not index event (status=%v): %v", status, msg)
			continue
		}

		debug("Failed to insert data(%v): %v", i, events[i])
		logp.Info("Bulk item insert failed (i=%v, status=%v): %v", i, status, msg)
		failed = append(failed, events[i])
	}
	return failed
}

func (proc *ProcessesWatcher) Init(config ProcsConfig) error {

	proc.proc_prefix = ""
	proc.PortProcMap = make(map[uint16]PortProcMapping)
	proc.LastMapUpdate = time.Now()

	proc.ReadFromProc = config.Enabled
	if proc.ReadFromProc {
		if runtime.GOOS != "linux" {
			proc.ReadFromProc = false
			logp.Info("Disabled /proc/ reading because not on linux")
		} else {
			logp.Info("Process matching enabled")
		}
	}

	if config.Max_proc_read_freq == 0 {
		proc.MaxReadFreq = 10 * time.Millisecond
	} else {
		proc.MaxReadFreq = time.Duration(config.Max_proc_read_freq) *
			time.Millisecond
	}

	if config.Refresh_pids_freq == 0 {
		proc.RefreshPidsFreq = 1 * time.Second
	} else {
		proc.RefreshPidsFreq = time.Duration(config.Refresh_pids_freq) *
			time.Millisecond
	}

	// Read the local IP addresses
	var err error
	proc.LocalAddrs, err = common.LocalIpAddrs()
	if err != nil {
		logp.Err("Error getting local IP addresses: %s", err)
		proc.LocalAddrs = []net.IP{}
	}

	if proc.ReadFromProc {
		for _, procConfig := range config.Monitored {

			grepper := procConfig.Cmdline_grep
			if len(grepper) == 0 {
				grepper = procConfig.Process
			}

			p, err := NewProcess(proc, procConfig.Process, grepper, time.Tick(proc.RefreshPidsFreq))
			if err != nil {
				logp.Err("NewProcess: %s", err)
			} else {
				proc.Processes = append(proc.Processes, p)
			}
		}
	}

	return nil
}

// Initialize Elasticsearch as output
func (out *ElasticsearchOutput) Init(config outputs.MothershipConfig, topology_expire int) error {

	if len(config.Protocol) == 0 {
		config.Protocol = "http"
	}

	url := fmt.Sprintf("%s://%s:%d%s", config.Protocol, config.Host, config.Port, config.Path)

	con := NewElasticsearch(url, config.Username, config.Password)
	out.Conn = con

	if config.Index != "" {
		out.Index = config.Index
	} else {
		out.Index = "packetbeat"
	}

	out.TopologyExpire = 15000
	if topology_expire != 0 {
		out.TopologyExpire = topology_expire /*sec*/ * 1000 // millisec
	}

	out.FlushInterval = 1000 * time.Millisecond
	if config.Flush_interval != nil {
		out.FlushInterval = time.Duration(*config.Flush_interval) * time.Millisecond
	}
	out.BulkMaxSize = 10000
	if config.Bulk_size != nil {
		out.BulkMaxSize = *config.Bulk_size
	}

	err := out.EnableTTL()
	if err != nil {
		logp.Err("Fail to set _ttl mapping: %s", err)
		return err
	}

	out.sendingQueue = make(chan BulkMsg, 1000)
	go out.SendMessagesGoroutine()

	logp.Info("[ElasticsearchOutput] Using Elasticsearch %s", url)
	logp.Info("[ElasticsearchOutput] Using index pattern [%s-]YYYY.MM.DD", out.Index)
	logp.Info("[ElasticsearchOutput] Topology expires after %ds", out.TopologyExpire/1000)
	if out.FlushInterval > 0 {
		logp.Info("[ElasticsearchOutput] Insert events in batches. Flush interval is %s. Bulk size is %d.", out.FlushInterval, out.BulkMaxSize)
	} else {
		logp.Info("[ElasticsearchOutput] Insert events one by one. This might affect the performance of the shipper.")
	}

	return nil
}

// initOffset finds the current offset of the file and sets it in the harvester as position
func (h *Harvester) initOffset() {
	// get current offset in file
	offset, _ := h.file.Seek(0, os.SEEK_CUR)

	if h.Offset > 0 {
		logp.Info("harvester", "harvest: %q position:%d (offset snapshot:%d)", h.Path, h.Offset, offset)
	} else if cfg.CmdlineOptions.TailOnRotate {
		logp.Info("harvester", "harvest: (tailing) %q (offset snapshot:%d)", h.Path, offset)
	} else {
		logp.Info("harvester", "harvest: %q (offset snapshot:%d)", h.Path, offset)
	}

	h.Offset = offset
}

// publish is used to publish events using the configured protocol client.
// It provides general error handling and back off support used on failed
// send attempts. To be used by PublishEvent and PublishEvents.
// The send callback will try to progress sending traffic and returns kind of
// progress made in ok or resetFail. If ok is set to true, send finished
// processing events. If ok is false but resetFail is set, send was partially
// successful. If send was partially successful, the fail counter is reset thus up
// to maxAttempts send attempts without any progress might be executed.
func (s *SingleConnectionMode) publish(
	signaler outputs.Signaler,
	send func() (ok bool, resetFail bool),
) error {
	fails := 0
	var backoffCount uint
	var err error

	for !s.closed && (s.maxAttempts == 0 || fails < s.maxAttempts) {
		ok := false
		resetFail := false

		if err := s.connect(); err != nil {
			logp.Info("Connecting error publishing events (retrying): %s", err)
			goto sendFail
		}

		ok, resetFail = send()
		if !ok {
			goto sendFail
		}

		outputs.SignalCompleted(signaler)
		return nil

	sendFail:
		fails++
		if resetFail {
			fails = 0
		}
		if s.maxAttempts > 0 && fails == s.maxAttempts {
			// max number of attempts reached
			break
		}

		logp.Info("send fail")
		backoff := time.Duration(int64(s.waitRetry) * (1 << backoffCount))
		if backoff > s.maxWaitRetry {
			backoff = s.maxWaitRetry
		} else {
			backoffCount++
		}
		logp.Info("backoff retry: %v", backoff)
		time.Sleep(backoff)
	}

	outputs.SignalFailed(signaler, err)
	return nil
}

// SafeFileRotate safely rotates an existing file under path and replaces it with the tempfile
func SafeFileRotate(path, tempfile string) error {
	old := path + ".old"
	var e error

	if e = os.Rename(path, old); e != nil {
		logp.Info("rotate: rename of %s to %s - %s", path, old, e)
		return e
	}

	if e = os.Rename(tempfile, path); e != nil {
		logp.Info("rotate: rename of %s to %s - %s", tempfile, path, e)
		return e
	}
	return nil
}

func LoadGeoIPData(config Geoip) *libgeo.GeoIP {

	geoipPaths := []string{}

	if config.Paths != nil {
		geoipPaths = *config.Paths
	}
	if len(geoipPaths) == 0 {
		logp.Info("GeoIP disabled: No paths were set under output.geoip.paths")
		// disabled
		return nil
	}

	// look for the first existing path
	var geoipPath string
	for _, path := range geoipPaths {
		fi, err := os.Lstat(path)
		if err != nil {
			logp.Err("GeoIP path could not be loaded: %s", path)
			continue
		}

		if fi.Mode()&os.ModeSymlink == os.ModeSymlink {
			// follow symlink
			geoipPath, err = filepath.EvalSymlinks(path)
			if err != nil {
				logp.Warn("Could not load GeoIP data: %s", err.Error())
				return nil
			}
		} else {
			geoipPath = path
		}
		break
	}

	if len(geoipPath) == 0 {
		logp.Warn("Couldn't load GeoIP database")
		return nil
	}

	geoLite, err := libgeo.Load(geoipPath)
	if err != nil {
		logp.Warn("Could not load GeoIP data: %s", err.Error())
	}

	logp.Info("Loaded GeoIP data from: %s", geoipPath)
	return geoLite
}

func (r *Registrar) Init() error {
	// Init state
	r.Persist = make(chan *FileState)
	r.State = make(map[string]*FileState)
	r.Channel = make(chan []*FileEvent, 1)

	// Set to default in case it is not set
	if r.registryFile == "" {
		r.registryFile = cfg.DefaultRegistryFile
	}

	// Make sure the directory where we store the registryFile exists
	absPath, err := filepath.Abs(r.registryFile)
	if err != nil {
		return fmt.Errorf("Failed to get the absolute path of %s: %v",
			r.registryFile, err)
	}
	r.registryFile = absPath

	// Create directory if it does not already exist.
	registryPath := filepath.Dir(r.registryFile)
	err = os.MkdirAll(registryPath, 0755)
	if err != nil {
		return fmt.Errorf("Failed to created registry file dir %s: %v",
			registryPath, err)
	}

	logp.Info("Registry file set to: %s", r.registryFile)

	return nil
}

// SafeFileRotate safely rotates an existing file under path and replaces it with the tempfile
func SafeFileRotate(path, tempfile string) error {
	if e := os.Rename(tempfile, path); e != nil {
		logp.Info("registry rotate: rename of %s to %s - %s", tempfile, path, e)
		return e
	}
	return nil
}

func (h *Harvester) open() *os.File {
	// Special handling that "-" means to read from standard input
	if h.Path == "-" {
		h.file = os.Stdin
		return h.file
	}

	for {
		var err error
		h.file, err = os.Open(h.Path)

		if err != nil {
			// retry on failure.
			logp.Info("harvester", "Failed opening %s: %s", h.Path, err)
			time.Sleep(5 * time.Second)
		} else {
			break
		}
	}

	file := &File{
		File: h.file,
	}

	// Check we are not following a rabbit hole (symlinks, etc.)
	if !file.IsRegularFile() {
		panic(fmt.Errorf("Harvester file error"))
	}

	h.setFileOffset()

	return h.file
}

// Starts scanning through all the file paths and fetch the related files. Start a harvester for each file
func (p *Prospector) Run(spoolChan chan *input.FileEvent) {

	p.running = true

	logp.Info("Starting prospector of type: %v", p.ProspectorConfig.Harvester.InputType)
	switch p.ProspectorConfig.Harvester.InputType {
	case cfg.StdinInputType:
		p.stdinRun(spoolChan)
		return
	case cfg.LogInputType:
		p.logRun(spoolChan)
		return
	}

	logp.Info("Invalid prospector type: %v")
}

func (eb *Winlogbeat) Setup(b *beat.Beat) error {
	eb.beat = b
	eb.client = b.Events
	eb.done = make(chan struct{})

	var err error
	eb.checkpoint, err = checkpoint.NewCheckpoint(eb.config.Winlogbeat.RegistryFile, 10, 5*time.Second)
	if err != nil {
		return err
	}

	if eb.config.Winlogbeat.Metrics.BindAddress != "" {
		bindAddress := eb.config.Winlogbeat.Metrics.BindAddress
		sock, err := net.Listen("tcp", bindAddress)
		if err != nil {
			return err
		}
		go func() {
			logp.Info("Metrics hosted at http://%s/debug/vars", bindAddress)
			err := http.Serve(sock, nil)
			if err != nil {
				logp.Warn("Unable to launch HTTP service for metrics. %v", err)
				return
			}
		}()
	}

	return nil
}