<commit_before>/*
 * Copyright 2014 <NAME>.
 *
 * This file is a modification of code from Logstash Forwarder.
 * Copyright 2012-2013 <NAME> and contributors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package main

import (
  "log"
  "os"
  "path/filepath"
  "time"
)

const (
  Status_Ok     = iota
  Status_Resume = iota
  Status_Failed = iota
)

type ProspectorInfo struct {
  file           string
  identity       FileIdentity
  last_seen      uint32
  harvester_cb   chan int64
  harvester_stop chan interface{}
  status         int
  running        bool
  orphaned       bool
  finish_offset  int64
}

func NewProspectorInfoFromFileState(file string, filestate *FileState) *ProspectorInfo {
  return &ProspectorInfo{
    file:           file,
    identity:       filestate,
    harvester_cb:   make(chan int64, 1),
    harvester_stop: make(chan interface{}),
    status:         Status_Resume,
    finish_offset:  filestate.Offset,
  }
}

func NewProspectorInfoFromFileInfo(file string, fileinfo os.FileInfo) *ProspectorInfo {
  return &ProspectorInfo{
    file:           file,
    identity:       &FileInfo{fileinfo: fileinfo}, // fileinfo is nil for stdin
    harvester_cb:   make(chan int64, 1),
    harvester_stop: make(chan interface{}),
  }
}

func (pi *ProspectorInfo) IsRunning() bool {
  if !pi.running {
    return false
  }
  select {
  case pi.finish_offset = <-pi.harvester_cb:
    pi.running = false
  default:
  }
  return pi.running
}

func (pi *ProspectorInfo) ShutdownSignal() <-chan interface{} {
  return pi.harvester_stop
}

func (pi *ProspectorInfo) Stop() {
  if !pi.running {
    return
  }
  if pi.file == "-" {
    // Just in case someone started us outside a pipeline with stdin
    // to stop confusion at why we don't exit after Ctrl+C
    // There's no deadline on Stdin reads :-(
    log.Printf("Waiting for Stdin to close (EOF or Ctrl+D)\n")
  }
  close(pi.harvester_stop)
}

func (pi *ProspectorInfo) Wait() {
  if !pi.running {
    return
  }
  pi.finish_offset = <-pi.harvester_cb
  pi.running = false
}

func (pi *ProspectorInfo) Update(fileinfo os.FileInfo, iteration uint32) {
  // Allow identity to replace itself with a new identity (this allows a FileState to promote itself to a FileInfo)
  pi.identity.Update(fileinfo, &pi.identity)
  pi.last_seen = iteration
}

type Prospector struct {
  shutdown         *LogCourierShutdown
  generalconfig    *GeneralConfig
  fileconfigs      []FileConfig
  prospectorindex  map[string]*ProspectorInfo
  prospectors      map[*ProspectorInfo]*ProspectorInfo
  from_beginning   bool
  iteration        uint32
  lastscan         time.Time
  registrar_events []RegistrarEvent
}

func NewProspector(config *Config, shutdown *LogCourierShutdown) *Prospector {
  return &Prospector{
    shutdown: shutdown,
    generalconfig: &config.General,
    fileconfigs: config.Files,
    from_beginning: *from_beginning,
    registrar_events: make([]RegistrarEvent, 0),
  }
}

func (p *Prospector) Prospect(resume map[string]*ProspectorInfo, registrar *Registrar, output chan<- *FileEvent) {
  defer func() {
    p.shutdown.Done()
  }()

  // Connect to the registrar
  registrar_chan := registrar.Connect()

  // Pre-populate prospectors with what we had previously
  p.prospectorindex = resume
  p.prospectors = make(map[*ProspectorInfo]*ProspectorInfo)
  for _, v := range resume {
    p.prospectors[v] = v
  }

  // Handle any "-" (stdin) paths - but only once
  stdin_started := false
  for _, config := range p.fileconfigs {
    for i, path := range config.Paths {
      if path == "-" {
        if !stdin_started {
          // We need to check err - we cannot allow a nil stat
          stat, err := os.Stdin.Stat()
          if err != nil {
            log.Printf("stat(Stdin) failed: %s\n", err)
            continue
          }

          // Stdin is implicitly an orphaned fileinfo
          info := NewProspectorInfoFromFileInfo("-", stat)
          info.orphaned = true

          // Store the reference so we can shut it down later
          p.prospectors[info] = info

          // Start the harvester
          p.startHarvesterWithOffset(info, output, &config, 0)

          stdin_started = true
        }

        // Remove it from the file list
        config.Paths = append(config.Paths[:i], config.Paths[i+1:]...)
      }
    }
  }

ProspectLoop:
  for {

    newlastscan := time.Now()
    p.iteration++ // Overflow is allowed

    for _, config := range p.fileconfigs {
      for _, path := range config.Paths {
        // Scan - flag false so new files always start at beginning
        p.scan(path, &config, registrar_chan, output)
      }
    }

    // We only obey *from_beginning (which is stored in this) on startup,
    // afterwards we force from beginning
    p.from_beginning = true

    // Clear out entries that no longer exist and we've stopped harvesting
    for _, info := range p.prospectors {
      if info.IsRunning() {
        continue
      }
      if !info.orphaned {
        if info.last_seen >= p.iteration {
          continue
        }<commit_msg>Fix reference problem with fields configuration Fixes #9<commit_after>/*
 * Copyright 2014 <NAME>.
 *
 * This file is a modification of code from Logstash Forwarder.
 * Copyright 2012-2013 <NAME> and contributors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package main

import (
  "log"
  "os"
  "path/filepath"
  "time"
)

const (
  Status_Ok     = iota
  Status_Resume = iota
  Status_Failed = iota
)

type ProspectorInfo struct {
  file           string
  identity       FileIdentity
  last_seen      uint32
  harvester_cb   chan int64
  harvester_stop chan interface{}
  status         int
  running        bool
  orphaned       bool
  finish_offset  int64
}

func NewProspectorInfoFromFileState(file string, filestate *FileState) *ProspectorInfo {
  return &ProspectorInfo{
    file:           file,
    identity:       filestate,
    harvester_cb:   make(chan int64, 1),
    harvester_stop: make(chan interface{}),
    status:         Status_Resume,
    finish_offset:  filestate.Offset,
  }
}

func NewProspectorInfoFromFileInfo(file string, fileinfo os.FileInfo) *ProspectorInfo {
  return &ProspectorInfo{
    file:           file,
    identity:       &FileInfo{fileinfo: fileinfo}, // fileinfo is nil for stdin
    harvester_cb:   make(chan int64, 1),
    harvester_stop: make(chan interface{}),
  }
}

func (pi *ProspectorInfo) IsRunning() bool {
  if !pi.running {
    return false
  }
  select {
  case pi.finish_offset = <-pi.harvester_cb:
    pi.running = false
  default:
  }
  return pi.running
}

func (pi *ProspectorInfo) ShutdownSignal() <-chan interface{} {
  return pi.harvester_stop
}

func (pi *ProspectorInfo) Stop() {
  if !pi.running {
    return
  }
  if pi.file == "-" {
    // Just in case someone started us outside a pipeline with stdin
    // to stop confusion at why we don't exit after Ctrl+C
    // There's no deadline on Stdin reads :-(
    log.Printf("Waiting for Stdin to close (EOF or Ctrl+D)\n")
  }
  close(pi.harvester_stop)
}

func (pi *ProspectorInfo) Wait() {
  if !pi.running {
    return
  }
  pi.finish_offset = <-pi.harvester_cb
  pi.running = false
}

func (pi *ProspectorInfo) Update(fileinfo os.FileInfo, iteration uint32) {
  // Allow identity to replace itself with a new identity (this allows a FileState to promote itself to a FileInfo)
  pi.identity.Update(fileinfo, &pi.identity)
  pi.last_seen = iteration
}

type Prospector struct {
  shutdown         *LogCourierShutdown
  generalconfig    *GeneralConfig
  fileconfigs      []FileConfig
  prospectorindex  map[string]*ProspectorInfo
  prospectors      map[*ProspectorInfo]*ProspectorInfo
  from_beginning   bool
  iteration        uint32
  lastscan         time.Time
  registrar_events []RegistrarEvent
}

func NewProspector(config *Config, shutdown *LogCourierShutdown) *Prospector {
  return &Prospector{
    shutdown: shutdown,
    generalconfig: &config.General,
    fileconfigs: config.Files,
    from_beginning: *from_beginning,
    registrar_events: make([]RegistrarEvent, 0),
  }
}

func (p *Prospector) Prospect(resume map[string]*ProspectorInfo, registrar *Registrar, output chan<- *FileEvent) {
  defer func() {
    p.shutdown.Done()
  }()

  // Connect to the registrar
  registrar_chan := registrar.Connect()

  // Pre-populate prospectors with what we had previously
  p.prospectorindex = resume
  p.prospectors = make(map[*ProspectorInfo]*ProspectorInfo)
  for _, v := range resume {
    p.prospectors[v] = v
  }

  // Handle any "-" (stdin) paths - but only once
  stdin_started := false
  for config_k, config := range p.fileconfigs {
    for i, path := range config.Paths {
      if path == "-" {
        if !stdin_started {
          // We need to check err - we cannot allow a nil stat
          stat, err := os.Stdin.Stat()
          if err != nil {
            log.Printf("stat(Stdin) failed: %s\n", err)
            continue
          }

          // Stdin is implicitly an orphaned fileinfo
          info := NewProspectorInfoFromFileInfo("-", stat)
          info.orphaned = true

          // Store the reference so we can shut it down later
          p.prospectors[info] = info

          // Start the harvester
          p.startHarvesterWithOffset(info, output, &p.fileconfigs[config_k], 0)

          stdin_started = true
        }

        // Remove it from the file list
        config.Paths = append(config.Paths[:i], config.Paths[i+1:]...)
      }
    }
  }

ProspectLoop:
  for {

    newlastscan := time.Now()
    p.iteration++ // Overflow is allowed

    for config_k, config := range p.fileconfigs {
      for _, path := range config.Paths {
        // Scan - flag false so new files always start at beginning
        p.scan(path, &p.fileconfigs[config_k], registrar_chan, output)
      }
    }

    // We only obey *from_beginning (which is stored in this) on startup,
    // afterwards we force from beginning
    p.from_beginning = true

    // Clear out entries that no longer exist and we've stopped harvesting
    for _, info := range p.prospectors {
      if info.IsRunning() {
        continue
      }
      if !info.orphaned {
        if info.last_seen >= p.iteration {
          continue
        }
{'fraction_non_alphanumeric': 0.09373611728120836, 'fraction_numerical': 0.0063083074189249225, 'mean_word_length': 2.712401055408971, 'pattern_counts': {'":': 0, '<': 15, '<?xml version=': 0, '>': 9, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '494449', 'n_tokens_mistral': 3530, 'n_tokens_neox': 3434, 'n_words': 1065}