Commit 800737d2 authored by Frédéric Logier's avatar Frédéric Logier
Browse files

first commit

parent bc4d2c58
export GOPATH=$GOPATH:/home/user/sources/graylog_http_push
cd /home/user/sources/graylog_http_push/src
Linux :
go build -o ../linux/graylog_push graylog_http_push.go
windows :
32 bits :
GOOS=windows GOARCH=386 go build -o ../win/graylog_http_push_i386.exe graylog_http_push.go
64 bits :
GOOS=windows GOARCH=amd64 go build -o ../win/graylog_http_push_amd64.exe graylog_http_push.go
# create a windows service
sc create graylogpush binpath= "\"C:\Users\user\sources\graylog_http_push\win\graylog_http_push_i386.exe\" \"C:\Users\user\sources\graylog_http_push\win\graylog_http_push.toml\" --service" depend= Tcpip
# This is a TOML document.
# payload_host should be : Customer-software-typeoflog
# payload_level is an int :
# 0 Emergency emerg (panic) Système inutilisable.
# 1 Alert alert Une intervention immédiate est nécessaire.
# 2 Critical crit Erreur critique pour le système.
# 3 Error err (error) Erreur de fonctionnement.
# 4 Warning warn (warning) Avertissement (une erreur peut intervenir si aucune action n'est prise).
# 5 Notice notice Événement normal méritant d'être signalé.
# 6 Informational info Pour information.
# 7 Debugging debug Message de mise au point.
# watcher_type should be "event" or "loop"
# loop sert a lire toutes les 60 secondes le contenu du repertoire de log et a renvoyer les logs presents a graylog.
# event utilise les notifications système pour signaler un evenement dans le repertoire de log, par exemple la creation d'un fichier. Un json est créé et si le serveur graylog n'est pas joignable il est stocké dans une base locale boltdb
title = "TOML graylog agent config file"
[owner]
name = "fredix"
organization = "fredix"
dob = 2016-06-12T12:00:00Z
[boltdb]
dbfile="/home/sysadmin/graylogpush_bolt.db"
[graylog]
url = "graylog.exemple.com"
format = "gelf"
protocol = "http"
port = 443
[watchers]
[watchers.test]
watcher_type = "event"
name = "Watch test1"
directory_log = "/tmp/test1"
ext_file = ".push"
directory_archive = "/tmp/test/archives"
payload_host = "project1"
payload_level = 7
[watchers.test2]
watcher_type = "event"
name = "Watcher test2"
directory_log = "/tmp/test2"
ext_file = ".push"
directory_archive = "/tmp/test2/archives"
payload_host = "project2"
payload_level = 7
[watchers.test3]
watcher_type = "loop"
name = "Watcher test3"
directory_log = "/tmp/test3"
ext_file = ".push"
directory_archive = "/tmp/test2/archives"
payload_host = "project3"
payload_level = 7
package gputils
import (
// standard packages
"bytes"
"encoding/binary"
"encoding/json"
"fmt"
"io"
"io/ioutil"
"log"
"net/http"
"os"
"path/filepath"
"strconv"
"strings"
"time"
// external packages
"github.com/boltdb/bolt"
"gopkg.in/fsnotify.v1"
)
type Message struct {
ID int
Version string
Message string `json:"short_message"`
Host string `json:"host"`
Level int `json:"level"`
MessageLog string `json:"_log"`
File string `json:"_file"`
ArchiveDir string `json:"_archivedir"`
Localtime string `json:"_localtime"`
}
type BoltDb struct {
Dbfile string `toml:"dbfile"`
DB *bolt.DB
writerChan chan [3]interface{}
graylog Graylog
}
type Graylog struct {
Url string `toml:"url"`
Port int `toml:"port"`
Format string `toml:"format"`
Protocol string `toml:"protocol"`
}
type Dir_watcher struct {
Watcher_type string
Name string
Directory_log string
Ext_file string
Directory_archive string
Payload_host string
Payload_level int
}
func LoopDirectoryLog(boltdb *BoltDb, graylog *Graylog, dir_watcher Dir_watcher) {
go func() {
for {
ListFilesAndPush(boltdb, graylog, dir_watcher)
time.Sleep(60000 * time.Millisecond)
}
}()
}
func (this *BoltDb) Resend(bucket string) {
go func() {
for {
this.DB.View(func(tx *bolt.Tx) error {
b := tx.Bucket([]byte(bucket))
if b == nil {
return nil
}
b.ForEach(func(k, v []byte) error {
//fmt.Printf("key=%d, value=%s\n", k, v)
//find a payload
payload := this.findOne(bucket, k)
// delete payload from boltdb
this.purgeOne(bucket, k)
// resend
fmt.Printf("Resend findone Message : %d %s %s %s %s\n", payload.ID, payload.Message, payload.Host, payload.File, payload.MessageLog)
this.pushPayload(payload)
return nil
})
//time.Sleep(5000 * time.Millisecond)
return nil
})
time.Sleep(15000 * time.Millisecond)
}
}()
}
func (this *BoltDb) pushPayload(payload *Message) {
fmt.Printf("pushPayload : %s\n", payload)
url := this.graylog.Protocol + "://" + this.graylog.Url + ":" + strconv.Itoa(this.graylog.Port) + "/" + this.graylog.Format
fmt.Printf("PushPayload graylog : %s\n", url)
go PushToGraylogHttp(
"event",
this,
url,
payload)
}
func (this *BoltDb) findOne(bucket string, key []byte) *Message {
var payload Message
this.DB.View(func(tx *bolt.Tx) error {
b := tx.Bucket([]byte(bucket))
if b == nil {
return nil
}
v := b.Get(key)
// fmt.Printf("findone data : %s\n", v)
if err := json.Unmarshal(v, &payload); err != nil {
return nil
}
return nil
})
return &payload
}
func (this *BoltDb) Purge(bucket string) {
this.DB.Update(func(tx *bolt.Tx) error {
b := tx.Bucket([]byte(bucket))
if b == nil {
return nil
}
b.ForEach(func(k, v []byte) error {
fmt.Printf("key=%s, value=%s\n", k, v)
b.Delete(k)
return nil
})
return nil
})
}
func (this *BoltDb) purgeOne(bucket string, key []byte) {
this.DB.Update(func(tx *bolt.Tx) error {
b := tx.Bucket([]byte(bucket))
if b == nil {
return nil
}
b.Delete(key)
return nil
})
}
func (this *BoltDb) writer() {
for data := range this.writerChan {
bucket := data[0].(string)
keyId := data[1].(string)
data := data[2].(*Message)
err := this.DB.Update(func(tx *bolt.Tx) error {
sessionBucket, err := tx.CreateBucketIfNotExists([]byte(bucket))
if err != nil {
return fmt.Errorf("create bucket: %s", err)
}
if keyId == "" {
id, _ := sessionBucket.NextSequence()
data.ID = int(id)
fmt.Println("payload ID : ", data.ID)
// Marshal user data into bytes.
buf, err := json.Marshal(data)
if err != nil {
return err
}
return sessionBucket.Put(itob(data.ID), buf)
} else {
// Marshal user data into bytes.
buf, err := json.Marshal(data)
if err != nil {
return err
}
return sessionBucket.Put([]byte(keyId), buf)
}
})
if err != nil {
// TODO: Handle instead of panic
panic(err)
}
}
}
func NewBoltDb(dbPath string, graylogconf Graylog) *BoltDb {
db, err := bolt.Open(dbPath, 0666, nil)
writerChan := make(chan [3]interface{})
boltDb := &BoltDb{DB: db, writerChan: writerChan, graylog: graylogconf}
go boltDb.writer()
if err != nil {
panic(err)
}
return boltDb
}
func cp(dst, src string) error {
s, err := os.Open(src)
if err != nil {
return err
}
// no need to check errors on read only file, we already got everything
// we need from the filesystem, so nothing can go wrong now.
defer s.Close()
d, err := os.Create(dst)
if err != nil {
return err
}
if _, err := io.Copy(d, s); err != nil {
d.Close()
return err
}
return d.Close()
}
func ListFilesAndPush(boltdb *BoltDb, graylog *Graylog, dir_watcher Dir_watcher) {
url := graylog.Protocol + "://" + graylog.Url + ":" + strconv.Itoa(graylog.Port) + "/" + graylog.Format
files, _ := ioutil.ReadDir(dir_watcher.Directory_log)
for _, f := range files {
file_ext := filepath.Ext(f.Name())
if file_ext == dir_watcher.Ext_file {
fmt.Println("file to push : ", dir_watcher.Directory_log+"/"+f.Name())
data, err := ioutil.ReadFile(dir_watcher.Directory_log + "/" + f.Name())
if err != nil {
panic(err)
}
log.Println("data: ", string(data))
payload := payload(dir_watcher.Directory_archive, dir_watcher.Name, string(data), dir_watcher.Directory_log+"/"+f.Name(), dir_watcher.Payload_host, dir_watcher.Payload_level)
go PushToGraylogHttp(
dir_watcher.Watcher_type,
boltdb,
url,
&payload)
}
}
}
func payload(archive_dir string, msg string, messagelog string, file string, host string, level int) Message {
t := time.Now()
m := Message{
Version: "1.1",
Message: msg,
Host: host,
Level: level,
MessageLog: messagelog,
File: file,
ArchiveDir: archive_dir,
Localtime: t.Format("01-02-2006T15-04-05"),
}
return m
}
func PushToGraylogHttp(watcher_type string, boltdb *BoltDb, url string, payload *Message) {
payload_id := payload.ID
archive := payload.ArchiveDir
file := payload.File
file_name := filepath.Base(file)
log.Println("file name :", file_name)
//url := "http://192.168.51.57:12201/gelf"
fmt.Println("URL:>", url)
fmt.Println("PAYLOAD ID:>", payload_id)
jsonStr, _ := json.Marshal(payload)
fmt.Println("json: ", string(jsonStr))
req, err := http.NewRequest("POST", url, bytes.NewBuffer(jsonStr))
// req.Header.Set("X-Custom-Header", "myvalue")
// req.Header.Set("Content-Type", "application/json")
client := &http.Client{}
resp, err := client.Do(req)
if err != nil {
//panic(err)
//log.Fatal("err")
log.Println("graylog post err :", err)
if watcher_type == "event" {
boltdb.writerChan <- [3]interface{}{"graylog", "", payload}
}
return
}
defer resp.Body.Close()
fmt.Println("response Status:", resp.Status)
fmt.Println("response Headers:", resp.Header)
body, _ := ioutil.ReadAll(resp.Body)
fmt.Println("response Body:", string(body))
if strings.Contains(resp.Status, "202") == true {
log.Println("status: ", resp.Status)
t := time.Now()
file_timestamp := t.Format("01-02-2006T15-04-05") + "-" + strconv.Itoa(t.Nanosecond())
file_name += "_" + file_timestamp
//fmt.Println("timestamp : ", file_timestamp)
err := cp(archive+"/"+file_name+".txt", file)
if err != nil {
log.Fatal(err)
} else {
os.Remove(file)
}
} else {
log.Fatal("Graylog server error : ", resp.Status)
// store payload to boltdb to send it later
//storeDB(dbfile, &m)
if watcher_type == "event" {
boltdb.writerChan <- [3]interface{}{"graylog", "", payload}
}
}
}
func LogNewWatcher(boltdb *BoltDb, graylog *Graylog, dir_watcher Dir_watcher) {
fmt.Println("watched dir :", dir_watcher.Directory_log)
watcher, err := fsnotify.NewWatcher()
if err != nil {
log.Fatal(err)
}
defer watcher.Close()
os.Mkdir(dir_watcher.Directory_archive, 0700)
done := make(chan bool)
go func() {
for {
select {
case event := <-watcher.Events:
log.Println("event:", event)
if event.Op&fsnotify.Create == fsnotify.Create {
log.Println("created file:", event.Name)
file_ext := filepath.Ext(event.Name)
if file_ext == dir_watcher.Ext_file {
data, err := ioutil.ReadFile(event.Name)
if err != nil {
panic(err)
}
log.Println("data: ", string(data))
//var sem = make(chan int, MaxTaches)
url := graylog.Protocol + "://" + graylog.Url + ":" + strconv.Itoa(graylog.Port) + "/" + graylog.Format
// fmt.Println("URL : " + url)
payload := payload(dir_watcher.Directory_archive, dir_watcher.Name, string(data), event.Name, dir_watcher.Payload_host, dir_watcher.Payload_level)
go PushToGraylogHttp(
dir_watcher.Watcher_type,
boltdb,
url,
&payload)
}
}
case err := <-watcher.Errors:
log.Println("error:", err)
}
}
}()
err = watcher.Add(dir_watcher.Directory_log)
if err != nil {
log.Fatal(err)
}
<-done
}
// not used
func storeDB(dbfile string, m *Message) {
db, err := bolt.Open(dbfile, 0600, nil)
if err != nil {
log.Fatal(err)
}
defer db.Close()
// store json to db
db.Update(func(tx *bolt.Tx) error {
b, err := tx.CreateBucketIfNotExists([]byte("graylog"))
if err != nil {
return fmt.Errorf("create bucket: %s", err)
}
id, _ := b.NextSequence()
m.ID = int(id)
// Marshal user data into bytes.
buf, err := json.Marshal(m)
if err != nil {
return err
}
fmt.Println("payload ID : ", m.ID)
// Persist bytes to users bucket.
return b.Put(itob(m.ID), buf)
//return b.Put([]byte("test"), []byte("My New Year post"))
})
}
// itob returns an 8-byte big endian representation of v.
func itob(v int) []byte {
b := make([]byte, 8)
binary.BigEndian.PutUint64(b, uint64(v))
return b
}
package main
import (
// internal package
"gputils"
// standard packages
"fmt"
"log"
"os"
"time"
// external packages
"github.com/BurntSushi/toml"
"github.com/kardianos/service"
)
// Program structures.
// Define Start and Stop methods.
type program struct {
exit chan struct{}
}
type tomlConfig struct {
Title string
Owner ownerInfo
Boltdb gputils.BoltDb
Graylog gputils.Graylog
Watchers map[string]gputils.Dir_watcher
}
type ownerInfo struct {
Name string
Org string `toml:"organization"`
DOB time.Time
}
var ConfigFile string = "graylog_http_push.toml"
var logger service.Logger
var MaxTaches int = 5
func (p *program) Start(s service.Service) error {
// Start should not block. Do the actual work async.
if service.Interactive() {
logger.Info("Running in terminal.")
} else {
logger.Info("Running under service manager.")
}
p.exit = make(chan struct{})
go p.run()
return nil
}
func (p *program) run() {
// Do work here
var config tomlConfig
if _, err := toml.DecodeFile(ConfigFile, &config); err != nil {
panic(fmt.Sprintf("%s", err))
}
fmt.Println("Title: ", config.Title)
fmt.Printf("Owner: %s (%s, %s)\n",
config.Owner.Name, config.Owner.Org, config.Owner.DOB)
fmt.Printf("Graylog: %s %d (Format %s) (Protocol %s)\n",
config.Graylog.Url, config.Graylog.Port, config.Graylog.Format,
config.Graylog.Protocol)
fmt.Println("Watchers")
// boltdb to store json payload
// and maybe other stuffs
boltDb := gputils.NewBoltDb(config.Boltdb.Dbfile, config.Graylog)
fmt.Println("list bolt")
boltDb.Purge("graylog")
boltDb.Resend("graylog")
// done := make(chan bool)
for watcherName, dir_watcher := range config.Watchers {
fmt.Printf("Watcher: %s (%s, %s, %s %s %d)\n", watcherName, dir_watcher.Directory_log, dir_watcher.Ext_file, dir_watcher.Directory_archive, dir_watcher.Payload_host, dir_watcher.Payload_level)
fmt.Println("watcher type : ", dir_watcher.Watcher_type)
if dir_watcher.Watcher_type == "event" {
fmt.Println("watcher type EVENT : ", dir_watcher.Watcher_type)
// list all files to push on startup
gputils.ListFilesAndPush(boltDb, &config.Graylog, dir_watcher)
// launch watchers
go gputils.LogNewWatcher(boltDb, &config.Graylog, dir_watcher)
} else if dir_watcher.Watcher_type == "loop" {