<commit_before>package handler

import (
	"encoding/json"
	"errors"
	"fmt"
	"github.com/streadway/amqp"
	"koding/db/models"
	"koding/db/mongodb"
	"koding/db/mongodb/modelhelper"
	"koding/kontrol/kontroldaemon/workerconfig"
	"koding/kontrol/kontrolhelper"
	"koding/tools/slog"
	"labix.org/v2/mgo"
	"labix.org/v2/mgo/bson"
	"strconv"
	"strings"
	"time"
)

type IncomingMessage struct {
	Worker  *models.Worker
	Monitor *models.Monitor
}

var producer *kontrolhelper.Producer

func Startup() {
	var err error
	producer, err = kontrolhelper.CreateProducer("worker")
	if err != nil {
		slog.Println(err)
	}

	err = producer.Channel.ExchangeDeclare("clientExchange", "fanout", true, false, false, false, nil)
	if err != nil {
		slog.Printf("clientExchange exchange.declare: %s\n", err)
	}

	runHelperFunctions()

	slog.Println("handler is initialized")
}

// runHelperFunctions contains several indepenendent helper functions that do
// certain tasks.
func runHelperFunctions() {
	// HeartBeat checker for workers
	tickerWorker := time.NewTicker(workerconfig.HEARTBEAT_INTERVAL)
	go func() {
		queryFunc := func(c *mgo.Collection) error {
			worker := models.Worker{}
			iter := c.Find(nil).Iter()
			for iter.Next(&worker) {
				if worker.Status == models.Dead {
					continue // already dead, nothing to do
				}

				if time.Now().Before(worker.Timestamp.Add(workerconfig.HEARTBEAT_DELAY)) {
					continue // still alive, pick up the next one
				}

				slog.Printf("[%s (%d)] no activity at '%s' - '%s' (pid: %d). marking them as dead\n",
					worker.Name,
					worker.Version,
					worker.Hostname,
					worker.Uuid,
					worker.Pid,
				)

				worker.Status = models.Dead
				worker.Monitor.Mem = models.MemData{}
				worker.Monitor.Uptime = 0
				modelhelper.UpdateIDWorker(worker)
			}

			if err := iter.Close(); err != nil {
				return err
			}

			return nil
		}

		for _ = range tickerWorker.C {
			mongodb.Run("jKontrolWorkers", queryFunc)
		}
	}()

	// Cleanup dead deployments at intervals. This goroutine will lookup at
	// each information if a deployment has running workers. If workers for a
	// certain deployment is not running anymore, then it will remove the
	// deployment information and all workers associated with that deployment
	// build.
	tickerDeployment := time.NewTicker(time.Hour * 1)
	go func() {
		for _ = range tickerDeployment.C {
			slog.Println("cleaner started to remove unused deployments and dead workers")
			infos := modelhelper.GetClients()
			for _, info := range infos {
				version, _ := strconv.Atoi(info.BuildNumber)

				// look if any workers are running for a certain version
				foundWorker := false
				query := func(c *mgo.Collection) error {
					iter := c.Find(bson.M{"version": version, "status": int(models.Started)}).Iter()
					worker := models.Worker{}
					for iter.Next(&worker) {
						foundWorker = true
					}

					if err := iter.Close(); err != nil {
						return err
					}

					return nil
				}

				mongodb.Run("jKontrolWorkers", query)

				// ... if not remove deployment information and dead workers of that version
				if !foundWorker {
					slog.Printf("removing deployment info for build number %s\n", info.BuildNumber)
					err := modelhelper.DeleteClient(info.BuildNumber)
					if err != nil {
						slog.Println(err)
					}

					slog.Printf("removing dead workers for build number %s\n", info.BuildNumber)
					query := func(c *mgo.Collection) error {
						_, err := c.RemoveAll(bson.M{"version": version, "status": int(models.Dead)})
						return err
					}

					err = mongodb.Run("jKontrolWorkers", query)
					if err != nil {
						slog.Println(err)
					}
				}
			}
		}
	}()
}

func ClientMessage(data amqp.Delivery) {
	if data.RoutingKey == "kontrol-client" {
		var info models.ServerInfo
		err := json.Unmarshal(data.Body, &info)
		if err != nil {
			slog.Printf("bad json client msg: %s\n", err)
		}

		modelhelper.AddClient(info)
	}
}

func WorkerMessage(data []byte) {
	var msg IncomingMessage
	err := json.Unmarshal(data, &msg)
	if err != nil {
		slog.Printf("bad json incoming msg: %s\n", err)
	}

	if msg.Monitor != nil {
		err := SaveMonitorData(msg.Monitor)
		if err != nil {
			slog.Println(err)
		}
	} else if msg.Worker != nil {
		err = DoWorkerCommand(msg.Worker.Message.Command, *msg.Worker)
		if err != nil {
			slog.Println(err)
		}
	} else {
		slog.Println("incoming message is in wrong format")
	}
}

func ApiMessage(data []byte) {
	var req workerconfig.ApiRequest
	err := json.Unmarshal(data, &req)
	if err != nil {
		slog.Printf("bad json incoming msg: %s\n", err)
	}

	err = DoApiRequest(req.Command, req.Uuid)
	if err != nil {
		slog.Println(err)<commit_msg>kontroldaemon: let error message more explicit for incoming data<commit_after>package handler

import (
	"encoding/json"
	"errors"
	"fmt"
	"github.com/streadway/amqp"
	"koding/db/models"
	"koding/db/mongodb"
	"koding/db/mongodb/modelhelper"
	"koding/kontrol/kontroldaemon/workerconfig"
	"koding/kontrol/kontrolhelper"
	"koding/tools/slog"
	"labix.org/v2/mgo"
	"labix.org/v2/mgo/bson"
	"strconv"
	"strings"
	"time"
)

type IncomingMessage struct {
	Worker  *models.Worker
	Monitor *models.Monitor
}

var producer *kontrolhelper.Producer

func Startup() {
	var err error
	producer, err = kontrolhelper.CreateProducer("worker")
	if err != nil {
		slog.Println(err)
	}

	err = producer.Channel.ExchangeDeclare("clientExchange", "fanout", true, false, false, false, nil)
	if err != nil {
		slog.Printf("clientExchange exchange.declare: %s\n", err)
	}

	runHelperFunctions()

	slog.Println("handler is initialized")
}

// runHelperFunctions contains several indepenendent helper functions that do
// certain tasks.
func runHelperFunctions() {
	// HeartBeat checker for workers
	tickerWorker := time.NewTicker(workerconfig.HEARTBEAT_INTERVAL)
	go func() {
		queryFunc := func(c *mgo.Collection) error {
			worker := models.Worker{}
			iter := c.Find(nil).Iter()
			for iter.Next(&worker) {
				if worker.Status == models.Dead {
					continue // already dead, nothing to do
				}

				if time.Now().Before(worker.Timestamp.Add(workerconfig.HEARTBEAT_DELAY)) {
					continue // still alive, pick up the next one
				}

				slog.Printf("[%s (%d)] no activity at '%s' - '%s' (pid: %d). marking them as dead\n",
					worker.Name,
					worker.Version,
					worker.Hostname,
					worker.Uuid,
					worker.Pid,
				)

				worker.Status = models.Dead
				worker.Monitor.Mem = models.MemData{}
				worker.Monitor.Uptime = 0
				modelhelper.UpdateIDWorker(worker)
			}

			if err := iter.Close(); err != nil {
				return err
			}

			return nil
		}

		for _ = range tickerWorker.C {
			mongodb.Run("jKontrolWorkers", queryFunc)
		}
	}()

	// Cleanup dead deployments at intervals. This goroutine will lookup at
	// each information if a deployment has running workers. If workers for a
	// certain deployment is not running anymore, then it will remove the
	// deployment information and all workers associated with that deployment
	// build.
	tickerDeployment := time.NewTicker(time.Hour * 1)
	go func() {
		for _ = range tickerDeployment.C {
			slog.Println("cleaner started to remove unused deployments and dead workers")
			infos := modelhelper.GetClients()
			for _, info := range infos {
				version, _ := strconv.Atoi(info.BuildNumber)

				// look if any workers are running for a certain version
				foundWorker := false
				query := func(c *mgo.Collection) error {
					iter := c.Find(bson.M{"version": version, "status": int(models.Started)}).Iter()
					worker := models.Worker{}
					for iter.Next(&worker) {
						foundWorker = true
					}

					if err := iter.Close(); err != nil {
						return err
					}

					return nil
				}

				mongodb.Run("jKontrolWorkers", query)

				// ... if not remove deployment information and dead workers of that version
				if !foundWorker {
					slog.Printf("removing deployment info for build number %s\n", info.BuildNumber)
					err := modelhelper.DeleteClient(info.BuildNumber)
					if err != nil {
						slog.Println(err)
					}

					slog.Printf("removing dead workers for build number %s\n", info.BuildNumber)
					query := func(c *mgo.Collection) error {
						_, err := c.RemoveAll(bson.M{"version": version, "status": int(models.Dead)})
						return err
					}

					err = mongodb.Run("jKontrolWorkers", query)
					if err != nil {
						slog.Println(err)
					}
				}
			}
		}
	}()
}

func ClientMessage(data amqp.Delivery) {
	if data.RoutingKey == "kontrol-client" {
		var info models.ServerInfo
		err := json.Unmarshal(data.Body, &info)
		if err != nil {
			slog.Printf("bad json client msg: %s err: %s\n", string(data.Body), err)
		}

		modelhelper.AddClient(info)
	}
}

func WorkerMessage(data []byte) {
	var msg IncomingMessage
	err := json.Unmarshal(data, &msg)
	if err != nil {
		slog.Printf("bad json incoming msg: %s err: %s\n", string(data), err)
	}

	if msg.Monitor != nil {
		err := SaveMonitorData(msg.Monitor)
		if err != nil {
			slog.Println(err)
		}
	} else if msg.Worker != nil {
		err = DoWorkerCommand(msg.Worker.Message.Command, *msg.Worker)
		if err != nil {
			slog.Println(err)
		}
	} else {
		slog.Println("incoming message is in wrong format")
	}
}

func ApiMessage(data []byte) {
	var req workerconfig.ApiRequest
	err := json.Unmarshal(data, &req)
	if err != nil {
		slog.Printf("bad json api msg: %s err: %s\n", string(data), err)
	}

	err = DoApiRequest(req.Command, req.Uuid)
	if err != nil {
		slog.Println(err)
{'fraction_non_alphanumeric': 0.13046238493281134, 'fraction_numerical': 0.0008464712728811766, 'mean_word_length': 3.232870577698164, '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': '2887301', 'n_tokens_mistral': 3749, 'n_tokens_neox': 3201, 'n_words': 812}