<commit_before>// Package agent runs readers, writers, and HTTP server.
package agent

import (
	"bytes"
	"encoding/json"
	"fmt"
	"net"
	"os"
	"strings"
	"time"

	"github.com/Sirupsen/logrus"
	resourced_config "github.com/resourced/resourced/config"
	resourced_executors "github.com/resourced/resourced/executors"
	resourced_host "github.com/resourced/resourced/host"
	"github.com/resourced/resourced/libprocess"
	"github.com/resourced/resourced/libstring"
	"github.com/resourced/resourced/libtime"
	resourced_readers "github.com/resourced/resourced/readers"
	"github.com/resourced/resourced/storage"
	resourced_writers "github.com/resourced/resourced/writers"
)

// NewAgent is the constructor for Agent struct.
func NewAgent(allowedNetworks []*net.IPNet) (*Agent, error) {
	agent := &Agent{}

	agent.AllowedNetworks = allowedNetworks

	agent.setTags()

	err := agent.setConfigStorage()
	if err != nil {
		return nil, err
	}

	agent.Db = storage.NewStorage()

	return agent, err
}

// Agent struct carries most of the functionality of ResourceD.
// It collects information through readers and serve them up as HTTP+JSON.
type Agent struct {
	ConfigStorage   *resourced_config.ConfigStorage
	DbPath          string
	Db              *storage.Storage
	Tags            []string
	AllowedNetworks []*net.IPNet
}

// setTags store RESOURCED_TAGS data to Tags field.
func (a *Agent) setTags() {
	a.Tags = make([]string, 0)

	tags := os.Getenv("RESOURCED_TAGS")
	if tags != "" {
		tagsSlice := strings.Split(tags, ",")
		a.Tags = make([]string, len(tagsSlice))

		for i, tag := range tagsSlice {
			a.Tags[i] = strings.TrimSpace(tag)
		}
	}
}

// pathWithPrefix prepends the short version of config.Kind to path.
func (a *Agent) pathWithPrefix(config resourced_config.Config) string {
	if config.Kind == "reader" {
		return a.pathWithKindPrefix("r", config)
	} else if config.Kind == "writer" {
		return a.pathWithKindPrefix("w", config)
	} else if config.Kind == "executor" {
		return a.pathWithKindPrefix("x", config)
	}
	return config.Path
}

// pathWithKindPrefix is common function called by pathWithReaderPrefix or pathWithWriterPrefix
func (a *Agent) pathWithKindPrefix(kind string, input interface{}) string {
	prefix := "/" + kind

	switch v := input.(type) {
	case resourced_config.Config:
		return prefix + v.Path
	case string:
		if strings.HasPrefix(v, prefix+"/") {
			return v
		} else {
			return prefix + v
		}
	}
	return ""
}

// Run executes a reader/writer config.
// Run will save reader data as JSON in local db.
func (a *Agent) Run(config resourced_config.Config) (output []byte, err error) {
	if config.Command != "" {
		output, err = a.runCommand(config)
	} else if config.GoStruct != "" && config.Kind == "reader" {
		output, err = a.runGoStructReader(config)
	} else if config.GoStruct != "" && config.Kind == "writer" {
		output, err = a.runGoStructWriter(config)
	} else if config.GoStruct != "" && config.Kind == "executor" {
		output, err = a.runGoStructExecutor(config)
		println(string(output))
	}

	if err != nil {
		logrus.WithFields(logrus.Fields{
			"Error":              err.Error(),
			"config.Command":     config.Command,
			"config.GoStruct":    config.GoStruct,
			"config.Path":        config.Path,
			"config.Interval":    config.Interval,
			"config.Kind":        config.Kind,
			"config.ReaderPaths": fmt.Sprintf("%s", config.ReaderPaths),
		}).Error("Failed to execute agent.runCommand/runGoStructReader/runGoStructWriter(config)")
	}

	err = a.saveRun(config, output, err)

	return output, err
}

// runCommand shells out external program and returns the output.
func (a *Agent) runCommand(config resourced_config.Config) ([]byte, error) {
	cmd := libprocess.NewCmd(config.Command)

	if config.Kind == "writer" {
		// Get readers data.
		readersData := make(map[string]interface{})

		for _, readerPath := range config.ReaderPaths {
			readerJsonBytes, err := a.GetRunByPath(a.pathWithKindPrefix("r", readerPath))

			if err == nil {
				var data interface{}
				err := json.Unmarshal(readerJsonBytes, &data)
				if err == nil {
					readersData[readerPath] = data
				}
			}
		}

		readersDataJsonBytes, err := json.Marshal(readersData)
		if err != nil {
			return nil, err
		}

		cmd.Stdin = bytes.NewReader(readersDataJsonBytes)
	}

	return cmd.Output()
}

// initGoStructReader initialize and return IReader.
func (a *Agent) initGoStructReader(config resourced_config.Config) (resourced_readers.IReader, error) {
	return resourced_readers.NewGoStructByConfig(config)
}

// initGoStructWriter initialize and return IWriter.
func (a *Agent) initGoStructWriter(config resourced_config.Config) (resourced_writers.IWriter, error) {
	writer, err := resourced_writers.NewGoStructByConfig(config)
	if err != nil {
		return nil, err
	}

	// Get readers data.
	readersData := make(map[string][]byte)

	for _, readerPath := range config.ReaderPaths {
		readerJsonBytes, err := a.GetRunByPath(a.pathWithKindPrefix("r", readerPath))
		if err == nil {
			readersData[readerPath] = readerJsonBytes
		}
	}

	writer.SetReadersDataInBytes(readersData)

	return writer, err
}

// initGoStructExecutor initialize and return IExecutor.
func (a *Agent) initGoStructExecutor(config resourced_config.Config) (resourced_executors.IExecutor, error) {
	executor, err := resourced_executors.NewGoStructByConfig(config)
	if err != nil {
		return nil, err
	}

	executor.SetReadersDataInBytes(a.Db.Data)

	return executor, nil
}

// runGoStruct executes IReader/IWriter and returns the output.
// Note that IWriter also implements IReader
func (a *Agent) runGoStruct(readerOrWriter resourced_readers.IReader) ([]byte, error) {
	err := readerOrWriter.Run()
	if err != nil {
		errData := make(map[string]string)
		errData["Error"] = err.Error()
		return json.Marshal(errData)
	}

	return readerOrWriter.ToJson()
}

// runGoStructReader executes IReader and returns the output.
func (a *Agent) runGoStructReader(config resourced_config.Config) ([]byte, error) {
	// Initialize IReader
	reader, err := a.initGoStructReader(config)
	if err != nil {
		return nil, err
	}

	return a.runGoStruct(reader)
}

// runGoStructWriter executes IWriter and returns error if exists.
func (a *Agent) runGoStructWriter(config resourced_config.Config) ([]byte, error) {
	// Initialize IWriter
	writer, err := a.initGoStructWriter(config)
	if err != nil {
		return nil, err
	}

	err = writer.GenerateData()
	if err != nil {
		logrus.WithFields(logrus.Fields{
			"Error":              err.Error(),
			"config.Command":     config.Command,
			"config.GoStruct":    config.GoStruct,
			"config.Path":        config.Path,
			"config.Interval":    config.Interval,
			"config.Kind":        config.Kind,
			"config.ReaderPaths": fmt.Sprintf("%s", config.ReaderPaths),
		}).Error("Failed to execute writer.GenerateData()")

		return nil, err
	}

	return a.runGoStruct(writer)
}

// runGoStructExecutor executes IExecutor and returns the output.
func (a *Agent) runGoStructExecutor(config resourced_config.Config) ([]byte, error) {
	// Initialize IExecutor
	executor, err := a.initGoStructExecutor(config)
	if err != nil {
		println("initGoStructExecutor failed: " + err.Error())
		return nil, err
	}

	return a.runGoStruct(executor)
}

// commonData gathers common information for every reader and writer.
func (a *Agent) commonData(config resourced_config.Config) map[string]interface{} {
	record := make(map[string]interface{})
	record["UnixNano"] = time.Now().UnixNano()
	record["Path"] = config.Path

	if config.Interval == "" {
		config.Interval = "1m"
	}
	record["Interval"] = config.Interval

	if config.Command != "" {
		record["Command"] = config.Command
	}

	if config.GoStruct != "" {
		record["GoStruct"] = config.GoStruct
	}

	return record
}

// hostData builds host related information.
func (a *Agent) hostData() (*resourced_host.Host, error) {
	host, err := resourced_host.NewHostByHostname()
	if err != nil {
		return nil, err
	}

	host.Tags = a.Tags

	return host, nil
}

// saveRun gathers basic, host, and reader/witer information and save them into local storage.
func (a *Agent) saveRun(config resourced_config.Config, output []byte, err error) error {
	// Do not perform save if config.Path is empty.
	if config.Path == "" {
		return nil
	}

	record := a.commonData(config)

	host, err := a.hostData()
	if err != nil {
		return err
	}
	record["Host"] = host

	if err == nil {
		runData := new(interface{})
		err = json.Unmarshal(output, &runData)
		if err != nil {
			return err
		}
		record["Data"] = runData

	} else {
		errMap := make(map[string]string)
		errMap["Error"] = err.Error()
		record["Data"] = errMap
	}

	recordInJson, err := json.Marshal(record)
	if err != nil {
		return err
	}

	a.Db.Set(a.pathWithPrefix(config), recordInJson)

	return err
}

// GetRun returns the JSON data stored in local storage given Config struct.
func (a *Agent) GetRun(config resourced_config.Config) ([]byte, error) {
	return a.GetRunByPath(a.pathWithPrefix(config))
}

// GetRunByPath returns JSON data stored in local storage given path string.
func (a *Agent) GetRunByPath(path string) ([]byte, error) {
	return a.Db.Get(path), nil
}

// RunForever executes Run() in an infinite loop with a sleep of config.Interval.
func (a *Agent) RunForever(config resourced_config.Config) {
	go func(a *Agent, config resourced_config.Config) {
		for {
			a.Run(config)
			libtime.SleepString(config.Interval)
		}
	}(a, config)
}

// RunAllForever executes all readers & writers in an infinite loop.
func (a *Agent) RunAllForever() {
	for _, config := range a.ConfigStorage.Readers {
		a.RunForever(config)
	}
	for _, config := range a.ConfigStorage.Writers {
		a.RunForever(config)
	}
	for _, config := range a.ConfigStorage.Executors {
		a.RunForever(config)
	}
}

// Check if a given IP:PORT is part of an allowed CIDR
func (a *Agent) IsAllowed(address string) bool {
	// Allow all if we allowed networks is not set
	if len(a.AllowedNetworks) == 0 {
		return true
	}

	ip := libstring.GetIP(address)
	if ip == nil {
		return false
	}

	// Check if IP is in one of our allowed networks
	for _, network := range a.AllowedNetworks {
		if network.Contains(ip) {
			return true
		}
	}

	return false
}
<commit_msg>Completed the refactor of moving Command to Gostruct shell.<commit_after>// Package agent runs readers, writers, and HTTP server.
package agent

import (
	"encoding/json"
	"fmt"
	"net"
	"os"
	"strings"
	"time"

	"github.com/Sirupsen/logrus"
	resourced_config "github.com/resourced/resourced/config"
	resourced_executors "github.com/resourced/resourced/executors"
	resourced_host "github.com/resourced/resourced/host"
	"github.com/resourced/resourced/libstring"
	"github.com/resourced/resourced/libtime"
	resourced_readers "github.com/resourced/resourced/readers"
	"github.com/resourced/resourced/storage"
	resourced_writers "github.com/resourced/resourced/writers"
)

// NewAgent is the constructor for Agent struct.
func NewAgent(allowedNetworks []*net.IPNet) (*Agent, error) {
	agent := &Agent{}

	agent.AllowedNetworks = allowedNetworks

	agent.setTags()

	err := agent.setConfigStorage()
	if err != nil {
		return nil, err
	}

	agent.Db = storage.NewStorage()

	return agent, err
}

// Agent struct carries most of the functionality of ResourceD.
// It collects information through readers and serve them up as HTTP+JSON.
type Agent struct {
	ConfigStorage   *resourced_config.ConfigStorage
	DbPath          string
	Db              *storage.Storage
	Tags            []string
	AllowedNetworks []*net.IPNet
}

// setTags store RESOURCED_TAGS data to Tags field.
func (a *Agent) setTags() {
	a.Tags = make([]string, 0)

	tags := os.Getenv("RESOURCED_TAGS")
	if tags != "" {
		tagsSlice := strings.Split(tags, ",")
		a.Tags = make([]string, len(tagsSlice))

		for i, tag := range tagsSlice {
			a.Tags[i] = strings.TrimSpace(tag)
		}
	}
}

// pathWithPrefix prepends the short version of config.Kind to path.
func (a *Agent) pathWithPrefix(config resourced_config.Config) string {
	if config.Kind == "reader" {
		return a.pathWithKindPrefix("r", config)
	} else if config.Kind == "writer" {
		return a.pathWithKindPrefix("w", config)
	} else if config.Kind == "executor" {
		return a.pathWithKindPrefix("x", config)
	}
	return config.Path
}

// pathWithKindPrefix is common function called by pathWithReaderPrefix or pathWithWriterPrefix
func (a *Agent) pathWithKindPrefix(kind string, input interface{}) string {
	prefix := "/" + kind

	switch v := input.(type) {
	case resourced_config.Config:
		return prefix + v.Path
	case string:
		if strings.HasPrefix(v, prefix+"/") {
			return v
		} else {
			return prefix + v
		}
	}
	return ""
}

// Run executes a reader/writer config.
// Run will save reader data as JSON in local db.
func (a *Agent) Run(config resourced_config.Config) (output []byte, err error) {
	if config.GoStruct != "" && config.Kind == "reader" {
		output, err = a.runGoStructReader(config)
	} else if config.GoStruct != "" && config.Kind == "writer" {
		output, err = a.runGoStructWriter(config)
	} else if config.GoStruct != "" && config.Kind == "executor" {
		output, err = a.runGoStructExecutor(config)
	}

	if err != nil {
		logrus.WithFields(logrus.Fields{
			"Error":              err.Error(),
			"config.Command":     config.Command,
			"config.GoStruct":    config.GoStruct,
			"config.Path":        config.Path,
			"config.Interval":    config.Interval,
			"config.Kind":        config.Kind,
			"config.ReaderPaths": fmt.Sprintf("%s", config.ReaderPaths),
		}).Error("Failed to execute runGoStructReader/runGoStructWriter/runGoStructExecutor")
	}

	err = a.saveRun(config, output, err)

	return output, err
}

// initGoStructReader initialize and return IReader.
func (a *Agent) initGoStructReader(config resourced_config.Config) (resourced_readers.IReader, error) {
	return resourced_readers.NewGoStructByConfig(config)
}

// initGoStructWriter initialize and return IWriter.
func (a *Agent) initGoStructWriter(config resourced_config.Config) (resourced_writers.IWriter, error) {
	writer, err := resourced_writers.NewGoStructByConfig(config)
	if err != nil {
		return nil, err
	}

	// Get readers data.
	readersData := make(map[string][]byte)

	for _, readerPath := range config.ReaderPaths {
		readerJsonBytes, err := a.GetRunByPath(a.pathWithKindPrefix("r", readerPath))
		if err == nil {
			readersData[readerPath] = readerJsonBytes
		}
	}

	writer.SetReadersDataInBytes(readersData)

	return writer, err
}

// initGoStructExecutor initialize and return IExecutor.
func (a *Agent) initGoStructExecutor(config resourced_config.Config) (resourced_executors.IExecutor, error) {
	executor, err := resourced_executors.NewGoStructByConfig(config)
	if err != nil {
		return nil, err
	}

	executor.SetReadersDataInBytes(a.Db.Data)

	return executor, nil
}

// runGoStruct executes IReader/IWriter and returns the output.
// Note that IWriter also implements IReader
func (a *Agent) runGoStruct(readerOrWriter resourced_readers.IReader) ([]byte, error) {
	err := readerOrWriter.Run()
	if err != nil {
		errData := make(map[string]string)
		errData["Error"] = err.Error()
		return json.Marshal(errData)
	}

	return readerOrWriter.ToJson()
}

// runGoStructReader executes IReader and returns the output.
func (a *Agent) runGoStructReader(config resourced_config.Config) ([]byte, error) {
	// Initialize IReader
	reader, err := a.initGoStructReader(config)
	if err != nil {
		return nil, err
	}

	return a.runGoStruct(reader)
}

// runGoStructWriter executes IWriter and returns error if exists.
func (a *Agent) runGoStructWriter(config resourced_config.Config) ([]byte, error) {
	// Initialize IWriter
	writer, err := a.initGoStructWriter(config)
	if err != nil {
		return nil, err
	}

	err = writer.GenerateData()
	if err != nil {
		logrus.WithFields(logrus.Fields{
			"Error":              err.Error(),
			"config.Command":     config.Command,
			"config.GoStruct":    config.GoStruct,
			"config.Path":        config.Path,
			"config.Interval":    config.Interval,
			"config.Kind":        config.Kind,
			"config.ReaderPaths": fmt.Sprintf("%s", config.ReaderPaths),
		}).Error("Failed to execute writer.GenerateData()")

		return nil, err
	}

	return a.runGoStruct(writer)
}

// runGoStructExecutor executes IExecutor and returns the output.
func (a *Agent) runGoStructExecutor(config resourced_config.Config) ([]byte, error) {
	// Initialize IExecutor
	executor, err := a.initGoStructExecutor(config)
	if err != nil {
		println("initGoStructExecutor failed: " + err.Error())
		return nil, err
	}

	return a.runGoStruct(executor)
}

// commonData gathers common information for every reader and writer.
func (a *Agent) commonData(config resourced_config.Config) map[string]interface{} {
	record := make(map[string]interface{})
	record["UnixNano"] = time.Now().UnixNano()
	record["Path"] = config.Path

	if config.Interval == "" {
		config.Interval = "1m"
	}
	record["Interval"] = config.Interval

	if config.Command != "" {
		record["Command"] = config.Command
	}

	if config.GoStruct != "" {
		record["GoStruct"] = config.GoStruct
	}

	return record
}

// hostData builds host related information.
func (a *Agent) hostData() (*resourced_host.Host, error) {
	host, err := resourced_host.NewHostByHostname()
	if err != nil {
		return nil, err
	}

	host.Tags = a.Tags

	return host, nil
}

// saveRun gathers basic, host, and reader/witer information and save them into local storage.
func (a *Agent) saveRun(config resourced_config.Config, output []byte, err error) error {
	// Do not perform save if config.Path is empty.
	if config.Path == "" {
		return nil
	}

	record := a.commonData(config)

	host, err := a.hostData()
	if err != nil {
		return err
	}
	record["Host"] = host

	if err == nil {
		runData := new(interface{})
		err = json.Unmarshal(output, &runData)
		if err != nil {
			return err
		}
		record["Data"] = runData

	} else {
		errMap := make(map[string]string)
		errMap["Error"] = err.Error()
		record["Data"] = errMap
	}

	recordInJson, err := json.Marshal(record)
	if err != nil {
		return err
	}

	a.Db.Set(a.pathWithPrefix(config), recordInJson)

	return err
}

// GetRun returns the JSON data stored in local storage given Config struct.
func (a *Agent) GetRun(config resourced_config.Config) ([]byte, error) {
	return a.GetRunByPath(a.pathWithPrefix(config))
}

// GetRunByPath returns JSON data stored in local storage given path string.
func (a *Agent) GetRunByPath(path string) ([]byte, error) {
	return a.Db.Get(path), nil
}

// RunForever executes Run() in an infinite loop with a sleep of config.Interval.
func (a *Agent) RunForever(config resourced_config.Config) {
	go func(a *Agent, config resourced_config.Config) {
		for {
			a.Run(config)
			libtime.SleepString(config.Interval)
		}
	}(a, config)
}

// RunAllForever executes all readers & writers in an infinite loop.
func (a *Agent) RunAllForever() {
	for _, config := range a.ConfigStorage.Readers {
		a.RunForever(config)
	}
	for _, config := range a.ConfigStorage.Writers {
		a.RunForever(config)
	}
	for _, config := range a.ConfigStorage.Executors {
		a.RunForever(config)
	}
}

// Check if a given IP:PORT is part of an allowed CIDR
func (a *Agent) IsAllowed(address string) bool {
	// Allow all if we allowed networks is not set
	if len(a.AllowedNetworks) == 0 {
		return true
	}

	ip := libstring.GetIP(address)
	if ip == nil {
		return false
	}

	// Check if IP is in one of our allowed networks
	for _, network := range a.AllowedNetworks {
		if network.Contains(ip) {
			return true
		}
	}

	return false
}

{'fraction_non_alphanumeric': 0.12608139237266444, 'fraction_numerical': 0.0003071410289224469, 'mean_word_length': 4.016949152542373, 'pattern_counts': {'":': 28, '<': 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': '5472248', 'n_tokens_mistral': 6819, 'n_tokens_neox': 6377, 'n_words': 1686}