<commit_before>// TODO
// * syntax check mig.Action.Arguments before exec()

package main

import (
	"encoding/json"
	"flag"
	"fmt"
	"github.com/streadway/amqp"
	"log"
	"mig"
	"mig/modules/filechecker"
	"os"
	"os/exec"
	"runtime"
	"time"
)

var AMQPBROKER string = "amqp://guest:guest@172.21.1.1:5672/"
var HEARTBEATFREQ string = "600s"

func main() {
	// parse command line argument
	// -m selects the mode {agent, filechecker, ...}
	var mode = flag.String("m", "agent",
		"module to run (eg. agent, filechecker)")
	flag.Parse()
	switch *mode {
	case "filechecker":
		// pass the rest of the arguments as a byte array
		// to the filechecker module
		var tmparg string
		for _, arg := range flag.Args() {
			tmparg = tmparg + arg
		}
		args := []byte(tmparg)
		fmt.Printf(filechecker.Run(args))
		os.Exit(0)
	case "agent":
		fmt.Printf("Launching MIG Agent")
	}

	// termChan is used to exit the program
	termChan := make(chan bool)
	actionsChan := make(chan []byte, 10)
	fCommandChan := make(chan mig.Command, 10)
	alertChan := make(chan mig.Alert, 10)
	resultChan := make(chan mig.Command, 10)
	hostname, err := os.Hostname()
	if err != nil {
		log.Fatalf("os.Hostname(): %v", err)
	}
	// declare a keepalive message to initiate registration
	regMsg := mig.KeepAlive{
		Name:          hostname,
		OS:            runtime.GOOS,
		QueueLoc:      fmt.Sprintf("%s.%s", runtime.GOOS, hostname),
		LastKeepAlive: time.Now(),
	}
	// define two bindings to receive msg from
	// mig.agt.<OS>.<hostname> is for agent specific messages
	// mig.all is for broadcasts
	agentQueue := fmt.Sprintf("mig.agt.%s", regMsg.QueueLoc)
	bindings := []mig.Binding{
		mig.Binding{agentQueue, agentQueue},
		mig.Binding{agentQueue, "mig.all"},
	}

	log.Println("MIG agent starting on", hostname)

	// Connects opens an AMQP connection from the credentials in the URL.
	conn, err := amqp.Dial(AMQPBROKER)
	if err != nil {
		log.Fatalf("amqp.Dial(): %v", err)
	}
	defer conn.Close()
	c, err := conn.Channel()
	if err != nil {
		log.Fatalf("conn.Channel(): %v", err)
	}
	for _, b := range bindings {
		_, err = c.QueueDeclare(b.Queue, // Queue name
			true,  // is durable
			false, // is autoDelete
			false, // is exclusive
			false, // is noWait
			nil)   // AMQP args
		if err != nil {
			log.Fatalf("QueueDeclare: %v", err)
		}
		err = c.QueueBind(b.Queue, // Queue name
			b.Key, // Routing key name
			"mig", // Exchange name
			false, // is noWait
			nil)   // AMQP args
		if err != nil {
			log.Fatalf("QueueBind: %v", err)
		}
	}

	// Limit the number of message the channel will receive
	err = c.Qos(2, // prefetch count (in # of msg)
		0,     // prefetch size (in bytes)
		false) // is global
	if err != nil {
		log.Fatalf("ChannelQoS: %v", err)
	}
	for _, b := range bindings {
		msgChan, err := c.Consume(b.Queue, // queue name
			"",    // some tag
			false, // is autoAck
			false, // is exclusive
			false, // is noLocal
			false, // is noWait
			nil)   // AMQP args
		if err != nil {
			log.Fatalf("ChannelConsume: %v", err)
		}
		go getCommands(msgChan, actionsChan, termChan)
	}
	go parseCommands(actionsChan, fCommandChan, termChan)
	go runFilechecker(fCommandChan, resultChan, termChan)
	go sendResults(c, regMsg.QueueLoc, resultChan, termChan)

	// All set, ready to keepAlive
	go keepAliveAgent(c, regMsg)

	// block until terminate chan is called
	<-termChan
}

// getCommands receives AMQP messages and pass them to the next level
func getCommands(messages <-chan amqp.Delivery, actions chan []byte, terminate chan bool) error {
	// range waits on the channel and returns all incoming messages
	// range will exit when the channel closes
	for m := range messages {
		log.Printf("getCommands: received '%s'", m.Body)
		// Ack this message only
		err := m.Ack(true)
		if err != nil {
			panic(err)
		}
		// pass it along to the parseCommands goroutine
		actions <- m.Body
		log.Printf("getCommands: queued in pos. %d", len(actions))
	}
	terminate <- true
	return nil
}<commit_msg>Agent: rearrange initialization in agent mode (need further squashing)<commit_after>// TODO
// * syntax check mig.Action.Arguments before exec()

package main

import (
	"encoding/json"
	"flag"
	"fmt"
	"github.com/streadway/amqp"
	"log"
	"mig"
	"mig/modules/filechecker"
	"os"
	"os/exec"
	"runtime"
	"time"
)

var AMQPBROKER string = "amqp://guest:guest@172.21.1.1:5672/"
var HEARTBEATFREQ string = "600s"

func main() {
	// parse command line argument
	// -m selects the mode {agent, filechecker, ...}
	var mode = flag.String("m", "agent",
		"module to run (eg. agent, filechecker)")
	flag.Parse()
	switch *mode {
	case "filechecker":
		// pass the rest of the arguments as a byte array
		// to the filechecker module
		var tmparg string
		for _, arg := range flag.Args() {
			tmparg = tmparg + arg
		}
		args := []byte(tmparg)
		fmt.Printf(filechecker.Run(args))
		os.Exit(0)
	case "agent":
		initAgent()
	}
}

// initAgent prepare the AMQP connections to the broker and launches the
// goroutines that will process commands received by the MIG Scheduler
func initAgent() (error){
	hostname, err := os.Hostname()
	log.Println("MIG agent starting on", hostname)
	// termChan is used to exit the program
	termChan := make(chan bool)
	actionsChan := make(chan []byte, 10)
	fCommandChan := make(chan mig.Command, 10)
	resultChan := make(chan mig.Command, 10)
	if err != nil {
		log.Fatalf("os.Hostname(): %v", err)
	}
	// declare a keepalive message to initiate registration
	regMsg := mig.KeepAlive{
		Name:          hostname,
		OS:            runtime.GOOS,
		QueueLoc:      fmt.Sprintf("%s.%s", runtime.GOOS, hostname),
		LastKeepAlive: time.Now(),
	}
	// define two bindings to receive msg from
	// mig.agt.<OS>.<hostname> is for agent specific messages
	// mig.all is for broadcasts
	agentQueue := fmt.Sprintf("mig.agt.%s", regMsg.QueueLoc)
	bindings := []mig.Binding{
		mig.Binding{agentQueue, agentQueue},
		mig.Binding{agentQueue, "mig.all"},
	}

	// Open an AMQP connection
	conn, err := amqp.Dial(AMQPBROKER)
	if err != nil {
		log.Fatalf("amqp.Dial(): %v", err)
	}
	defer conn.Close()
	c, err := conn.Channel()
	if err != nil {
		log.Fatalf("conn.Channel(): %v", err)
	}
	for _, b := range bindings {
		_, err = c.QueueDeclare(b.Queue, // Queue name
			true,  // is durable
			false, // is autoDelete
			false, // is exclusive
			false, // is noWait
			nil)   // AMQP args
		if err != nil {
			log.Fatalf("QueueDeclare: %v", err)
		}
		err = c.QueueBind(b.Queue, // Queue name
			b.Key, // Routing key name
			"mig", // Exchange name
			false, // is noWait
			nil)   // AMQP args
		if err != nil {
			log.Fatalf("QueueBind: %v", err)
		}
	}

	// Limit the number of message the channel will receive
	err = c.Qos(2, // prefetch count (in # of msg)
		0,     // prefetch size (in bytes)
		false) // is global
	if err != nil {
		log.Fatalf("ChannelQoS: %v", err)
	}
	for _, b := range bindings {
		msgChan, err := c.Consume(b.Queue, // queue name
			"",    // some tag
			false, // is autoAck
			false, // is exclusive
			false, // is noLocal
			false, // is noWait
			nil)   // AMQP args
		if err != nil {
			log.Fatalf("ChannelConsume: %v", err)
		}
		go getCommands(msgChan, actionsChan, termChan)
	}
	go parseCommands(actionsChan, fCommandChan, termChan)
	go runFilechecker(fCommandChan, resultChan, termChan)
	go sendResults(c, regMsg.QueueLoc, resultChan, termChan)

	// All set, ready to keepAlive
	go keepAliveAgent(c, regMsg)

	// block until terminate chan is called
	<-termChan
	return nil
}

// getCommands receives AMQP messages and pass them to the next level
func getCommands(messages <-chan amqp.Delivery, actions chan []byte, terminate chan bool) error {
	// range waits on the channel and returns all incoming messages
	// range will exit when the channel closes
	for m := range messages {
		log.Printf("getCommands: received '%s'", m.Body)
		// Ack this message only
		err := m.Ack(true)
		if err != nil {
			panic(err)
		}
		// pass it along to the parseCommands goroutine
		actions <- m.Body
		log.Printf("getCommands: queued in pos. %d", len(actions))
	}
	terminate <- true
	return nil
}
{'fraction_non_alphanumeric': 0.13906056860321384, 'fraction_numerical': 0.005933250927070457, 'mean_word_length': 3.1322778345250257, 'pattern_counts': {'":': 4, '<': 15, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 2, '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': '5358958', 'n_tokens_mistral': 3123, 'n_tokens_neox': 2874, 'n_words': 875}