<commit_before>package main

import (
	"encoding/json"
	"fmt"
	"github.com/ActiveState/log"
	"github.com/ActiveState/tail"
	"logyard"
	"os"
	"stackato/server"
	"unicode/utf8"
)

func tailLogFile(
	name string, filepath string,
	nodeid string, c *logyard.Client) (*tail.Tail, error) {
	if filepath == "" {
		filepath = fmt.Sprintf("/s/logs/%s.log", name)
	}

	log.Info("Tailing... ", filepath)

	t, err := tail.TailFile(filepath, tail.Config{
		MaxLineSize: Config.MaxRecordSize,
		MustExist:   false,
		Follow:      true,
		// ignore existing content, to support subsequent re-runs of systail
		Location: 0,
		ReOpen:   true,
		Poll:     false})

	if err != nil {
		return nil, err
	}

	go func(name string, tail *tail.Tail) {
		for line := range tail.Lines {
			// JSON must be a valid UTF-8 string
			if !utf8.ValidString(line.Text) {
				line.Text = string([]rune(line.Text))
			}
			data, err := json.Marshal(map[string]interface{}{
				"UnixTime": line.Time.Unix(),
				"Text":     line.Text,
				"Name":     name,
				"NodeID":   nodeid})
			if err != nil {
				tail.Killf("Failed to convert to JSON: %v", err)
				break
			}
			err = c.Send("systail."+name+"."+nodeid, string(data))
			if err != nil {
				log.Fatal("Failed to send to logyard: ", err)
			}
		}
	}(name, t)

	return t, nil
}

func main() {
	LoadConfig()

	nodeid, err := server.LocalIP()
	if err != nil {
		log.Fatalf("Failed to determine IP addr: %v", err)
	}
	log.Info("Host IP: ", nodeid)

	c, err := logyard.NewClientGlobal()
	if err != nil {
		log.Fatal(err)
	}
	tailers := []*tail.Tail{}

	fmt.Printf("%+v\n", Config.LogFiles)
	if len(Config.LogFiles) == 0 {
		log.Fatal("No log files configured in doozer")
	}

	for name, logfile := range Config.LogFiles {
		t, err := tailLogFile(name, logfile, nodeid, c)
		if err != nil {
			log.Fatal(err)
		}
		tailers = append(tailers, t)
	}

	for _, tail := range tailers {
		err := tail.Wait()
		if err != nil {
			log.Errorf("Cannot tail [%s]: %s", tail.Filename, err)
		}
	}

	// we don't expect any of the tailers to exit with or without
	// error.
	log.Error("No file left to tail; exiting.")
	os.Exit(1)
}
<commit_msg>systail: use zeromq sockets in thread-safe manner<commit_after>package main

import (
	"encoding/json"
	"fmt"
	"github.com/ActiveState/log"
	"github.com/ActiveState/tail"
	"github.com/alecthomas/gozmq"
	"logyard"
	"os"
	"stackato/server"
	"unicode/utf8"
)

func tailLogFile(name string, filepath string, nodeid string) (*tail.Tail, error) {
	if filepath == "" {
		filepath = fmt.Sprintf("/s/logs/%s.log", name)
	}

	log.Info("Tailing... ", filepath)

	t, err := tail.TailFile(filepath, tail.Config{
		MaxLineSize: Config.MaxRecordSize,
		MustExist:   false,
		Follow:      true,
		// ignore existing content, to support subsequent re-runs of systail
		Location: 0,
		ReOpen:   true,
		Poll:     false})

	if err != nil {
		return nil, err
	}

	go func(name string, tail *tail.Tail) {
		c, err := logyard.NewClientGlobal()
		if err != nil {
			log.Fatal(err)
		}
		defer c.Close()

		for line := range tail.Lines {
			// JSON must be a valid UTF-8 string
			if !utf8.ValidString(line.Text) {
				line.Text = string([]rune(line.Text))
			}
			data, err := json.Marshal(map[string]interface{}{
				"UnixTime": line.Time.Unix(),
				"Text":     line.Text,
				"Name":     name,
				"NodeID":   nodeid})
			if err != nil {
				tail.Killf("Failed to convert to JSON: %v", err)
				break
			}
			err = c.Send("systail."+name+"."+nodeid, string(data))
			if err != nil {
				log.Fatal("Failed to send to logyard: ", err)
			}
		}
	}(name, t)

	return t, nil
}

func main() {
	major, minor, patch := gozmq.Version()
	log.Infof("Starting systail (zeromq %d.%d.%d)", major, minor, patch)

	LoadConfig()

	nodeid, err := server.LocalIP()
	if err != nil {
		log.Fatalf("Failed to determine IP addr: %v", err)
	}
	log.Info("Host IP: ", nodeid)

	tailers := []*tail.Tail{}

	fmt.Printf("%+v\n", Config.LogFiles)
	if len(Config.LogFiles) == 0 {
		log.Fatal("No log files configured in doozer")
	}

	for name, logfile := range Config.LogFiles {
		t, err := tailLogFile(name, logfile, nodeid)
		if err != nil {
			log.Fatal(err)
		}
		tailers = append(tailers, t)
	}

	for _, tail := range tailers {
		err := tail.Wait()
		if err != nil {
			log.Errorf("Cannot tail [%s]: %s", tail.Filename, err)
		}
	}

	// we don't expect any of the tailers to exit with or without
	// error.
	log.Error("No file left to tail; exiting.")
	os.Exit(1)
}

{'fraction_non_alphanumeric': 0.16308243727598568, 'fraction_numerical': 0.002688172043010753, 'mean_word_length': 3.092575618698442, 'pattern_counts': {'":': 8, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, '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': '3174753', 'n_tokens_mistral': 1873, 'n_tokens_neox': 1687, 'n_words': 440}