나만의 Beats 만들기

package main

import (
	"fmt"
	"io/ioutil"
	"os"
)

func main() {
	//파라메터를 안 받았다면 현재 경로 "." 이하 파일, 디렉토리들 출력
	if len(os.Args) == 1 {
		listDir(".")
	} else {
		listDir(os.Args[1])
	}
}

func listDir(dirFile string) {
	files, _ := ioutil.ReadDir(dirFile)
	for _, f := range files {
		t := f.ModTime()
		fmt.Println(f.Name(), dirFile+"/"+f.Name(), f.IsDir(), t, f.Size()) //파일(디렉토리)명, 전체path, 디렉토리여부, 생성시간, 파일크기 출력
    
    //디렉토리인 경우 하위 파일들 재귀 호출.
		if f.IsDir() {
			listDir(dirFile + "/" + f.Name())
		}
	}
}

type LsbeatConfig struct {
	Period string `config:"period"`
	Path   string `config:"path"` //path 환경변수 값 추가.
}

type Beater interface {
	Config(*Beat) error  // Read and validate configuration.
	Setup(*Beat) error   // Initialize the Beat.
	Run(*Beat) error     // The main event loop. This method should block until signalled to stop by an invocation of the Stop() method.
	Cleanup(*Beat) error // Cleanup is invoked to perform any final clean-up prior to exiting.
	Stop()               // Stop is invoked to signal that the Run method should finish its execution. It will be invoked at most once.
}

type Lsbeat struct {
	beatConfig   *config.Config
	done         chan struct{}
	period       time.Duration
	client       publisher.Client
	path         string    //root 디렉토리
	lasIndexTime time.Time //가장 마지막에 검색한 시간
}

func (bt *Lsbeat) Setup(b *beat.Beat) error {

	// Setting default period if not set
	if bt.beatConfig.Lsbeat.Period == "" {
		bt.beatConfig.Lsbeat.Period = "1s"
	}
  
  // path 값을 lsbeat.yml 에 설정된 값으로 설정.
	if bt.beatConfig.Lsbeat.Path == "" {
		bt.beatConfig.Lsbeat.Path = "."
	}
	bt.path = bt.beatConfig.Lsbeat.Path
  
	bt.client = b.Publisher.Connect()

	var err error
	bt.period, err = time.ParseDuration(bt.beatConfig.Lsbeat.Period)
	if err != nil {
		return err
	}

	return nil
}

func (bt *Lsbeat) Run(b *beat.Beat) error {
	logp.Info("lsbeat is running! Hit CTRL-C to stop it.")

	ticker := time.NewTicker(bt.period)
	counter := 1
	for {
		select {
		case <-bt.done:
			return nil
		case <-ticker.C:
		}

		event := common.MapStr{
			"@timestamp": common.Time(time.Now()),
			"type":       b.Name,
			"counter":    counter,
		}
		bt.client.PublishEvent(event)
		logp.Info("Event sent")
		counter++
	}
}

import (
	"fmt"
	"io/ioutil" //listDir 에서 사용하기 위해 추가
	"time"
  
... 중략 ...

// 파일, 디렉토리 조회해서 출력하는 함수
func listDir(dirFile string) {
	files, _ := ioutil.ReadDir(dirFile)
	for _, f := range files {
		t := f.ModTime()
    fmt.Println(f.Name(), dirFile+"/"+f.Name(), f.IsDir(), t, f.Size())
    
		if f.IsDir() {
			listDir(dirFile+"/"+f.Name(), bt, b, counter)
		}
	}
}

// bt *Lsbeat, b *beat.Beat, counter int 추가
func listDir(dirFile string, bt *Lsbeat, b *beat.Beat, counter int) {
	files, _ := ioutil.ReadDir(dirFile)
	for _, f := range files {
		t := f.ModTime()
		//fmt.Println(f.Name(), dirFile+"/"+f.Name(), f.IsDir(), t, f.Size())

		event := common.MapStr{
			"@timestamp":  common.Time(time.Now()),
			"type":        b.Name,
			"counter":     counter,
			"modTime":     t,
			"filename":    f.Name(),
			"fullname":    dirFile + "/" + f.Name(),
			"isDirectory": f.IsDir(),
			"fileSize":    f.Size(),
		}
		//첫번째 실행인 경우 전체 파일 목록 색인.
		if counter == 1 {
			bt.client.PublishEvent(event) //elasticsearch 로 데이터 전송.
		} else {
			//2번째 이후 인 경우 지난번 실행 이후에 추가된 파일만 색인
			if t.After(bt.lasIndexTime) {
				bt.client.PublishEvent(event) //elasticsearch 로 데이터 전송.
			}
		}

		if f.IsDir() {
			listDir(dirFile+"/"+f.Name(), bt, b, counter) // bt *Lsbeat, b *beat.Beat, counter int 추가
		}
	}
}

func (bt *Lsbeat) Run(b *beat.Beat) error {
	logp.Info("lsbeat is running! Hit CTRL-C to stop it.")

	ticker := time.NewTicker(bt.period)
	counter := 1
	for {
		listDir(bt.path, bt, b, counter) // lsbeat 데이터 송신
		bt.lasIndexTime = time.Now()     // 송신 후 타임스탬프 기록.

		select {
		case <-bt.done:
			return nil
		case <-ticker.C:
		}

		// 기존 elasticsearch 전송 로직 삭제
		// event := common.MapStr{
		// 	"@timestamp": common.Time(time.Now()),
		// 	"type":       b.Name,
		// 	"counter":    counter,
		// }
		// bt.client.PublishEvent(event)

		logp.Info("Event sent")
		counter++
	}
}