<commit_before>package application

import (
	"encoding/json"
	"fmt"
	"net/http"
	"net/smtp"
	"os"
	"os/exec"
	"strings"
	"sync"

	"github.com/Sirupsen/logrus"
	"github.com/carbocation/interpose"
	"github.com/gorilla/csrf"
	"github.com/gorilla/mux"
	"github.com/gorilla/sessions"
	"github.com/justinas/alice"
	_ "github.com/lib/pq"
	"github.com/marcw/pagerduty"
	"github.com/resourced/resourced-master/config"
	"github.com/resourced/resourced-master/dal"
	"github.com/resourced/resourced-master/handlers"
	"github.com/resourced/resourced-master/libsmtp"
	"github.com/resourced/resourced-master/libstring"
	"github.com/resourced/resourced-master/libtime"
	"github.com/resourced/resourced-master/middlewares"
)

// New is the constructor for Application struct.
func New(configDir string) (*Application, error) {
	generalConfig, err := config.NewGeneralConfig(configDir)
	if err != nil {
		return nil, err
	}

	dbConfig, err := config.NewDBConfig(generalConfig)
	if err != nil {
		return nil, err
	}

	hostname, err := os.Hostname()
	if err != nil {
		return nil, err
	}

	app := &Application{}
	app.Hostname = hostname
	app.GeneralConfig = generalConfig
	app.DBConfig = dbConfig
	app.cookieStore = sessions.NewCookieStore([]byte(app.GeneralConfig.CookieSecret))
	app.csrfProtect = csrf.Protect([]byte(app.GeneralConfig.CookieSecret))

	return app, err
}

// Application is the application object that runs HTTP server.
type Application struct {
	Hostname      string
	GeneralConfig config.GeneralConfig
	DBConfig      *config.DBConfig
	cookieStore   *sessions.CookieStore
	csrfProtect   func(http.Handler) http.Handler
}

func (app *Application) MiddlewareStruct() (*interpose.Middleware, error) {
	middle := interpose.New()
	middle.Use(middlewares.SetAddr(app.GeneralConfig.Addr))
	middle.Use(middlewares.SetDBs(app.DBConfig))
	middle.Use(middlewares.SetCookieStore(app.cookieStore))

	middle.UseHandler(app.mux())

	return middle, nil
}

func (app *Application) mux() *mux.Router {
	MustLogin := middlewares.MustLogin
	MustLoginApi := middlewares.MustLoginApi
	SetClusters := middlewares.SetClusters

	router := mux.NewRouter()

	router.HandleFunc("/signup", handlers.GetSignup).Methods("GET")
	router.HandleFunc("/signup", handlers.PostSignup).Methods("POST")
	router.HandleFunc("/login", handlers.GetLogin).Methods("GET")
	router.HandleFunc("/login", handlers.PostLogin).Methods("POST")
	router.HandleFunc("/logout", handlers.GetLogout).Methods("GET")

	router.Handle("/", alice.New(MustLogin, SetClusters).ThenFunc(handlers.GetHosts)).Methods("GET")

	router.Handle("/saved-queries", alice.New(MustLogin).ThenFunc(handlers.PostSavedQueries)).Methods("POST")
	router.Handle("/saved-queries/{id:[0-9]+}", alice.New(MustLogin).ThenFunc(handlers.PostPutDeleteSavedQueriesID)).Methods("POST", "PUT", "DELETE")

	router.Handle("/graphs", alice.New(MustLogin, SetClusters).ThenFunc(handlers.GetGraphs)).Methods("GET")
	router.Handle("/graphs", alice.New(MustLogin, SetClusters).ThenFunc(handlers.PostGraphs)).Methods("POST")
	router.Handle("/graphs/{id:[0-9]+}", alice.New(MustLogin, SetClusters).ThenFunc(handlers.GetPostPutDeleteGraphsID)).Methods("GET", "POST", "PUT", "DELETE")

	router.Handle("/watchers", alice.New(MustLogin, SetClusters).ThenFunc(handlers.GetWatchers)).Methods("GET")
	router.Handle("/watchers", alice.New(MustLogin, SetClusters).ThenFunc(handlers.PostWatchers)).Methods("POST")
	router.Handle("/watchers/{id:[0-9]+}", alice.New(MustLogin, SetClusters).ThenFunc(handlers.PostPutDeleteWatcherID)).Methods("POST", "PUT", "DELETE")
	router.Handle("/watchers/{id:[0-9]+}/silence", alice.New(MustLogin, SetClusters).ThenFunc(handlers.PostWatcherIDSilence)).Methods("POST")

	router.Handle("/watchers/active", alice.New(MustLogin, SetClusters).ThenFunc(handlers.GetWatchersActive)).Methods("GET")
	router.Handle("/watchers/active", alice.New(MustLogin, SetClusters).ThenFunc(handlers.PostWatchersActive)).Methods("POST")
	router.Handle("/watchers/active/{id:[0-9]+}", alice.New(MustLogin, SetClusters).ThenFunc(handlers.PostPutDeleteWatcherActiveID)).Methods("POST", "PUT", "DELETE")

	router.Handle("/watchers/{watcherid:[0-9]+}/triggers", alice.New(MustLogin, SetClusters).ThenFunc(handlers.PostWatchersTriggers)).Methods("POST")
	router.Handle("/watchers/{watcherid:[0-9]+}/triggers/{id:[0-9]+}", alice.New(MustLogin, SetClusters).ThenFunc(handlers.PostPutDeleteWatcherTriggerID)).Methods("POST", "PUT", "DELETE")

	router.Handle("/metadata", alice.New(MustLogin, SetClusters).ThenFunc(handlers.GetMetadata)).Methods("GET")
	router.Handle("/metadata", alice.New(MustLogin, SetClusters).ThenFunc(handlers.PostMetadata)).Methods("POST")
	router.Handle(`/metadata/{key}`, alice.New(MustLogin, SetClusters).ThenFunc(handlers.DeleteMetadataKey)).Methods("POST", "DELETE")

	router.Handle("/users/{id:[0-9]+}", alice.New(MustLogin).ThenFunc(handlers.PostPutDeleteUsersID)).Methods("POST", "PUT", "DELETE")

	router.Handle("/clusters", alice.New(MustLogin, SetClusters).ThenFunc(handlers.GetClusters)).Methods("GET")
	router.Handle("/clusters", alice.New(MustLogin).ThenFunc(handlers.PostClusters)).Methods("POST")
	router.Handle("/clusters/{id:[0-9]+}", alice.New(MustLogin, SetClusters).ThenFunc(handlers.PostPutDeleteClusterID)).Methods("POST", "PUT", "DELETE")

	router.Handle("/clusters/current", alice.New(MustLogin, SetClusters).ThenFunc(handlers.PostClustersCurrent)).Methods("POST")

	router.Handle("/clusters/{id:[0-9]+}/access-tokens", alice.New(MustLogin).ThenFunc(handlers.PostAccessTokens)).Methods("POST")

	router.Handle("/clusters/{clusterid:[0-9]+}/metrics", alice.New(MustLogin).ThenFunc(handlers.PostMetrics)).Methods("POST")
	router.Handle("/clusters/{clusterid:[0-9]+}/metrics/{id:[0-9]+}", alice.New(MustLogin).ThenFunc(handlers.PostPutDeleteMetricID)).Methods("POST", "PUT", "DELETE")

	router.Handle("/access-tokens/{id:[0-9]+}/level", alice.New(MustLogin).ThenFunc(handlers.PostAccessTokensLevel)).Methods("POST")
	router.Handle("/access-tokens/{id:[0-9]+}/enabled", alice.New(MustLogin).ThenFunc(handlers.PostAccessTokensEnabled)).Methods("POST")

	router.Handle("/api/hosts", alice.New(MustLoginApi).ThenFunc(handlers.GetApiHosts)).Methods("GET")
	router.Handle("/api/hosts", alice.New(MustLoginApi).ThenFunc(handlers.PostApiHosts)).Methods("POST")

	router.Handle("/api/metadata", alice.New(MustLoginApi).ThenFunc(handlers.GetApiMetadata)).Methods("GET")
	router.Handle(`/api/metadata/{key}`, alice.New(MustLoginApi).ThenFunc(handlers.PostApiMetadataKey)).Methods("POST")
	router.Handle(`/api/metadata/{key}`, alice.New(MustLoginApi).ThenFunc(handlers.DeleteApiMetadataKey)).Methods("DELETE")
	router.Handle(`/api/metadata/{key}`, alice.New(MustLoginApi).ThenFunc(handlers.GetApiMetadataKey)).Methods("GET")

	router.Handle("/api/metrics/{id}/hosts/{host}", alice.New(MustLoginApi).ThenFunc(handlers.GetApiTSMetricsByHost)).Methods("GET")
	router.Handle("/api/metrics/{id}", alice.New(MustLoginApi).ThenFunc(handlers.GetApiTSMetrics)).Methods("GET")
	router.Handle("/api/metrics/{id}/15min", alice.New(MustLoginApi).ThenFunc(handlers.GetApiTSMetrics15Min)).Methods("GET")

	// Path of static files must be last!
	router.PathPrefix("/").Handler(http.FileServer(http.Dir("static")))

	return router
}

func (app *Application) PruneAll() {
	for {
		app.PruneTSWatchersOnce()
		app.PruneTSMetricsOnce()

		libtime.SleepString("24h")
	}
}

func (app *Application) PruneTSWatchersOnce() {
	for _, db := range app.DBConfig.TSWatchers.DBs {
		go func() {
			err := dal.NewTSWatcher(db).DeleteByDayInterval(nil, app.GeneralConfig.Watchers.DataRetention)
			if err != nil {
				logrus.WithFields(logrus.Fields{
					"Method":        "TSWatcher.DeleteByDayInterval",
					"DataRetention": app.GeneralConfig.Watchers.DataRetention,
				}).Error(err)
			}
		}()
	}
}

func (app *Application) PruneTSMetricsOnce() {
	for _, db := range app.DBConfig.TSMetrics.DBs {
		go func() {
			err := dal.NewTSMetric(db).DeleteByDayInterval(nil, app.GeneralConfig.Metrics.DataRetention)
			if err != nil {
				logrus.WithFields(logrus.Fields{
					"Method":        "TSMetric.DeleteByDayInterval",
					"DataRetention": app.GeneralConfig.Metrics.DataRetention,
				}).Error(err)
			}
		}()
	}
}

func (app *Application) WatchAll() {
	watcherRowsChan := make(chan []*dal.WatcherRow)

	// Fetch daemons and watchers data every 5 minutes
	go func() {
		for {
			daemonHostnames, _ := dal.NewDaemon(app.DBConfig.Core).AllHostnames(nil)
			groupedWatcherRows, _ := dal.NewWatcher(app.DBConfig.Core).AllSplitToDaemons(nil, daemonHostnames)
			watcherRowsChan <- groupedWatcherRows[app.Hostname]

			libtime.SleepString(app.GeneralConfig.Watchers.ListFetchInterval)
		}
	}()

	go func() {
		for {
			select {
			case watcherRows := <-watcherRowsChan:
				for _, watcherRow := range watcherRows {
					go func(watcherRow *dal.WatcherRow) {
						if watcherRow.IsPassive() {
							// Passive watching and triggering
							err := app.PassiveWatchOnce(watcherRow.ClusterID, watcherRow)
							if err != nil {
								logrus.WithFields(logrus.Fields{
									"Method":    "Application.PassiveWatchOnce",
									"ClusterID": watcherRow.ClusterID,
									"WatcherID": watcherRow.ID,
								}).Error(err)
							}

							err = app.RunTrigger(watcherRow.ClusterID, watcherRow)
							if err != nil {
								logrus.WithFields(logrus.Fields{
									"Method":    "Application.RunTrigger",
									"ClusterID": watcherRow.ClusterID,
									"WatcherID": watcherRow.ID,
								}).Error(err)
							}
						} else {
							// Active watching and triggering
							err := app.ActiveWatchOnce(watcherRow.ClusterID, watcherRow)
							if err != nil {
								logrus.WithFields(logrus.Fields{
									"Method":    "Application.ActiveWatchOnce",
									"ClusterID": watcherRow.ClusterID,
									"WatcherID": watcherRow.ID,
								}).Error(err)
							}
						}

						libtime.SleepString(watcherRow.CheckInterval)
					}(watcherRow)
				}
			}
		}
	}()
}

func (app *Application) PassiveWatchOnce(clusterID int64, watcherRow *dal.WatcherRow) error {
	affectedHosts, err := dal.NewHost(app.DBConfig.Core).AllByClusterIDQueryAndUpdatedInterval(nil, clusterID, watcherRow.SavedQuery.String, watcherRow.HostsLastUpdatedForPostgres())
	if err != nil {
		return err
	}

	numAffectedHosts := int64(len(affectedHosts))

	if numAffectedHosts == 0 || numAffectedHosts >= watcherRow.LowAffectedHosts {
		tsWatcherDataHosts := make([]string, numAffectedHosts)
		for i, affectedHost := range affectedHosts {
			tsWatcherDataHosts[i] = affectedHost.Name
		}

		tsWatcherData := make(map[string]interface{})
		tsWatcherData["hosts"] = tsWatcherDataHosts

		jsonData, err := json.Marshal(tsWatcherData)
		if err != nil {
			return err
		}

		// Write to ts_watchers asynchronously
		dbs := app.DBConfig.TSWatchers.PickMultipleForWrites()
		for _, db := range dbs {
			go func() {
				err := dal.NewTSWatcher(db).Create(nil, clusterID, watcherRow.ID, numAffectedHosts, jsonData)
				if err != nil {
					logrus.WithFields(logrus.Fields{
						"Method":          "TSWatcher.Create",
						"ClusterID":       clusterID,
						"WatcherID":       watcherRow.ID,
						"NumAffectedHost": numAffectedHosts,
					}).Error(err)
				}
			}()
		}
	}

	return err
}

func (app *Application) ActiveWatchOnce(clusterID int64, watcherRow *dal.WatcherRow) error {
	// Don't do anything if watcher is silenced.
	if watcherRow.IsSilenced {
		return nil
	}

	var hostnames []string

	// Checking hosts based on external lists.
	if len(watcherRow.HostsList()) > 0 {
		hostnames = watcherRow.HostsList()
	} else {
		hosts, err := dal.NewHost(app.DBConfig.Core).AllByClusterIDAndUpdatedInterval(nil, clusterID, watcherRow.HostsLastUpdatedForPostgres())
		if err != nil {
			return err
		}

		hostnames = make([]string, len(hosts))

		for i, host := range hosts {
			hostnames[i] = host.Name
		}
	}

	// If there are no hostnames, then we are not checking anything
	if len(hostnames) == 0 {
		return nil
	}

	numAffectedHosts := 0
	tsWatcherDataHosts := make([]string, 0)

	var errCollectorMutex sync.Mutex

	for _, hostname := range hostnames {
		if watcherRow.Command() == "ping" {
			go func(hostname string) {
				_, err := exec.Command("ping", "-c", "1", hostname).Output()
				if err != nil {
					errCollectorMutex.Lock()
					numAffectedHosts = numAffectedHosts + 1
					tsWatcherDataHosts = append(tsWatcherDataHosts, hostname)
					errCollectorMutex.Unlock()

					logrus.WithFields(logrus.Fields{
						"Method": "ping -c 1 " + hostname,
					}).Error(err)
				}
			}(hostname)

		} else if watcherRow.Command() == "ssh" {
			go func(hostname string) {
				sshOptions := []string{"-o BatchMode=yes", "-o ConnectTimeout=10"}

				if watcherRow.SSHPort() != "" {
					sshOptions = append(sshOptions, []string{"-p", watcherRow.SSHPort()}...)
				}

				userAtHost := hostname

				if watcherRow.SSHUser() != "" {
					userAtHost = fmt.Sprintf("%v@%v", watcherRow.SSHUser(), hostname)
				}

				sshOptions = append(sshOptions, userAtHost)

				outBytes, err := exec.Command("ssh", sshOptions...).CombinedOutput()
				outString := string(outBytes)

				// We only care about SSH connectivity
				if err != nil && !strings.Contains(outString, "Permission denied") && !strings.Contains(outString, "Host key verification failed") {
					errCollectorMutex.Lock()
					numAffectedHosts = numAffectedHosts + 1
					tsWatcherDataHosts = append(tsWatcherDataHosts, hostname)
					errCollectorMutex.Unlock()

					logrus.WithFields(logrus.Fields{
						"Method": "ssh -o BatchMode=yes -o ConnectTimeout=10 " + userAtHost,
					}).Error(err)
				}
			}(hostname)

		} else if watcherRow.Command() == "http" {
			go func(hostname string) {
				url := fmt.Sprintf("%v://%v:%s", watcherRow.HTTPScheme(), hostname, watcherRow.HTTPPort())

				client := &http.Client{}
				req, err := http.NewRequest(strings.ToUpper(watcherRow.HTTPMethod()), url, nil)
				if err != nil {
					logrus.WithFields(logrus.Fields{
						"URL":    url,
						"Method": "http.NewRequest",
					}).Error(err)
					return
				}

				if watcherRow.HTTPUser() != "" || watcherRow.HTTPPass() != "" {
					req.SetBasicAuth(watcherRow.HTTPUser(), watcherRow.HTTPPass())
				}

				resp, err := client.Do(req)
				if err != nil {
					logrus.WithFields(logrus.Fields{
						"URL":    url,
						"Method": "http.Client{}.Do",
					}).Error(err)
					return
				}

				if err != nil || (resp != nil && resp.StatusCode != watcherRow.HTTPCode()) {
					errCollectorMutex.Lock()
					numAffectedHosts = numAffectedHosts + 1
					tsWatcherDataHosts = append(tsWatcherDataHosts, hostname)
					errCollectorMutex.Unlock()

					logrus.WithFields(logrus.Fields{
						"URL":                url,
						"StatusCode":         resp.StatusCode,
						"ExpectedStatusCode": watcherRow.HTTPCode(),
						"Method":             "HTTP Client fetch",
					}).Error(err)
				}
			}(hostname)
		}
	}

	if numAffectedHosts == 0 || int64(numAffectedHosts) >= watcherRow.LowAffectedHosts {
		tsWatcherData := make(map[string]interface{})
		tsWatcherData["hosts"] = tsWatcherDataHosts

		jsonData, err := json.Marshal(tsWatcherData)
		if err != nil {
			return err
		}

		// Write to ts_watchers asynchronously
		dbs := app.DBConfig.TSWatchers.PickMultipleForWrites()
		for _, db := range dbs {
			go func() {
				err := dal.NewTSWatcher(db).Create(nil, clusterID, watcherRow.ID, int64(numAffectedHosts), jsonData)
				if err != nil {
					logrus.WithFields(logrus.Fields{
						"Method":          "TSWatcher.Create",
						"ClusterID":       clusterID,<commit_msg>Able to check agent using HEAD.<commit_after>package application

import (
	"encoding/json"
	"fmt"
	"net/http"
	"net/smtp"
	"os"
	"os/exec"
	"strings"
	"sync"

	"github.com/Sirupsen/logrus"
	"github.com/carbocation/interpose"
	"github.com/gorilla/csrf"
	"github.com/gorilla/mux"
	"github.com/gorilla/sessions"
	"github.com/justinas/alice"
	_ "github.com/lib/pq"
	"github.com/marcw/pagerduty"
	"github.com/resourced/resourced-master/config"
	"github.com/resourced/resourced-master/dal"
	"github.com/resourced/resourced-master/handlers"
	"github.com/resourced/resourced-master/libsmtp"
	"github.com/resourced/resourced-master/libstring"
	"github.com/resourced/resourced-master/libtime"
	"github.com/resourced/resourced-master/middlewares"
)

// New is the constructor for Application struct.
func New(configDir string) (*Application, error) {
	generalConfig, err := config.NewGeneralConfig(configDir)
	if err != nil {
		return nil, err
	}

	dbConfig, err := config.NewDBConfig(generalConfig)
	if err != nil {
		return nil, err
	}

	hostname, err := os.Hostname()
	if err != nil {
		return nil, err
	}

	app := &Application{}
	app.Hostname = hostname
	app.GeneralConfig = generalConfig
	app.DBConfig = dbConfig
	app.cookieStore = sessions.NewCookieStore([]byte(app.GeneralConfig.CookieSecret))
	app.csrfProtect = csrf.Protect([]byte(app.GeneralConfig.CookieSecret))

	return app, err
}

// Application is the application object that runs HTTP server.
type Application struct {
	Hostname      string
	GeneralConfig config.GeneralConfig
	DBConfig      *config.DBConfig
	cookieStore   *sessions.CookieStore
	csrfProtect   func(http.Handler) http.Handler
}

func (app *Application) MiddlewareStruct() (*interpose.Middleware, error) {
	middle := interpose.New()
	middle.Use(middlewares.SetAddr(app.GeneralConfig.Addr))
	middle.Use(middlewares.SetDBs(app.DBConfig))
	middle.Use(middlewares.SetCookieStore(app.cookieStore))

	middle.UseHandler(app.mux())

	return middle, nil
}

func (app *Application) mux() *mux.Router {
	MustLogin := middlewares.MustLogin
	MustLoginApi := middlewares.MustLoginApi
	SetClusters := middlewares.SetClusters

	router := mux.NewRouter()

	router.HandleFunc("/signup", handlers.GetSignup).Methods("GET")
	router.HandleFunc("/signup", handlers.PostSignup).Methods("POST")
	router.HandleFunc("/login", handlers.GetLogin).Methods("GET")
	router.HandleFunc("/login", handlers.PostLogin).Methods("POST")
	router.HandleFunc("/logout", handlers.GetLogout).Methods("GET")

	router.Handle("/", alice.New(MustLogin, SetClusters).ThenFunc(handlers.GetHosts)).Methods("GET")

	router.Handle("/saved-queries", alice.New(MustLogin).ThenFunc(handlers.PostSavedQueries)).Methods("POST")
	router.Handle("/saved-queries/{id:[0-9]+}", alice.New(MustLogin).ThenFunc(handlers.PostPutDeleteSavedQueriesID)).Methods("POST", "PUT", "DELETE")

	router.Handle("/graphs", alice.New(MustLogin, SetClusters).ThenFunc(handlers.GetGraphs)).Methods("GET")
	router.Handle("/graphs", alice.New(MustLogin, SetClusters).ThenFunc(handlers.PostGraphs)).Methods("POST")
	router.Handle("/graphs/{id:[0-9]+}", alice.New(MustLogin, SetClusters).ThenFunc(handlers.GetPostPutDeleteGraphsID)).Methods("GET", "POST", "PUT", "DELETE")

	router.Handle("/watchers", alice.New(MustLogin, SetClusters).ThenFunc(handlers.GetWatchers)).Methods("GET")
	router.Handle("/watchers", alice.New(MustLogin, SetClusters).ThenFunc(handlers.PostWatchers)).Methods("POST")
	router.Handle("/watchers/{id:[0-9]+}", alice.New(MustLogin, SetClusters).ThenFunc(handlers.PostPutDeleteWatcherID)).Methods("POST", "PUT", "DELETE")
	router.Handle("/watchers/{id:[0-9]+}/silence", alice.New(MustLogin, SetClusters).ThenFunc(handlers.PostWatcherIDSilence)).Methods("POST")

	router.Handle("/watchers/active", alice.New(MustLogin, SetClusters).ThenFunc(handlers.GetWatchersActive)).Methods("GET")
	router.Handle("/watchers/active", alice.New(MustLogin, SetClusters).ThenFunc(handlers.PostWatchersActive)).Methods("POST")
	router.Handle("/watchers/active/{id:[0-9]+}", alice.New(MustLogin, SetClusters).ThenFunc(handlers.PostPutDeleteWatcherActiveID)).Methods("POST", "PUT", "DELETE")

	router.Handle("/watchers/{watcherid:[0-9]+}/triggers", alice.New(MustLogin, SetClusters).ThenFunc(handlers.PostWatchersTriggers)).Methods("POST")
	router.Handle("/watchers/{watcherid:[0-9]+}/triggers/{id:[0-9]+}", alice.New(MustLogin, SetClusters).ThenFunc(handlers.PostPutDeleteWatcherTriggerID)).Methods("POST", "PUT", "DELETE")

	router.Handle("/metadata", alice.New(MustLogin, SetClusters).ThenFunc(handlers.GetMetadata)).Methods("GET")
	router.Handle("/metadata", alice.New(MustLogin, SetClusters).ThenFunc(handlers.PostMetadata)).Methods("POST")
	router.Handle(`/metadata/{key}`, alice.New(MustLogin, SetClusters).ThenFunc(handlers.DeleteMetadataKey)).Methods("POST", "DELETE")

	router.Handle("/users/{id:[0-9]+}", alice.New(MustLogin).ThenFunc(handlers.PostPutDeleteUsersID)).Methods("POST", "PUT", "DELETE")

	router.Handle("/clusters", alice.New(MustLogin, SetClusters).ThenFunc(handlers.GetClusters)).Methods("GET")
	router.Handle("/clusters", alice.New(MustLogin).ThenFunc(handlers.PostClusters)).Methods("POST")
	router.Handle("/clusters/{id:[0-9]+}", alice.New(MustLogin, SetClusters).ThenFunc(handlers.PostPutDeleteClusterID)).Methods("POST", "PUT", "DELETE")

	router.Handle("/clusters/current", alice.New(MustLogin, SetClusters).ThenFunc(handlers.PostClustersCurrent)).Methods("POST")

	router.Handle("/clusters/{id:[0-9]+}/access-tokens", alice.New(MustLogin).ThenFunc(handlers.PostAccessTokens)).Methods("POST")

	router.Handle("/clusters/{clusterid:[0-9]+}/metrics", alice.New(MustLogin).ThenFunc(handlers.PostMetrics)).Methods("POST")
	router.Handle("/clusters/{clusterid:[0-9]+}/metrics/{id:[0-9]+}", alice.New(MustLogin).ThenFunc(handlers.PostPutDeleteMetricID)).Methods("POST", "PUT", "DELETE")

	router.Handle("/access-tokens/{id:[0-9]+}/level", alice.New(MustLogin).ThenFunc(handlers.PostAccessTokensLevel)).Methods("POST")
	router.Handle("/access-tokens/{id:[0-9]+}/enabled", alice.New(MustLogin).ThenFunc(handlers.PostAccessTokensEnabled)).Methods("POST")

	router.Handle("/api/hosts", alice.New(MustLoginApi).ThenFunc(handlers.GetApiHosts)).Methods("GET")
	router.Handle("/api/hosts", alice.New(MustLoginApi).ThenFunc(handlers.PostApiHosts)).Methods("POST")

	router.Handle("/api/metadata", alice.New(MustLoginApi).ThenFunc(handlers.GetApiMetadata)).Methods("GET")
	router.Handle(`/api/metadata/{key}`, alice.New(MustLoginApi).ThenFunc(handlers.PostApiMetadataKey)).Methods("POST")
	router.Handle(`/api/metadata/{key}`, alice.New(MustLoginApi).ThenFunc(handlers.DeleteApiMetadataKey)).Methods("DELETE")
	router.Handle(`/api/metadata/{key}`, alice.New(MustLoginApi).ThenFunc(handlers.GetApiMetadataKey)).Methods("GET")

	router.Handle("/api/metrics/{id}/hosts/{host}", alice.New(MustLoginApi).ThenFunc(handlers.GetApiTSMetricsByHost)).Methods("GET")
	router.Handle("/api/metrics/{id}", alice.New(MustLoginApi).ThenFunc(handlers.GetApiTSMetrics)).Methods("GET")
	router.Handle("/api/metrics/{id}/15min", alice.New(MustLoginApi).ThenFunc(handlers.GetApiTSMetrics15Min)).Methods("GET")

	// Path of static files must be last!
	router.PathPrefix("/").Handler(http.FileServer(http.Dir("static")))

	return router
}

func (app *Application) PruneAll() {
	for {
		app.PruneTSWatchersOnce()
		app.PruneTSMetricsOnce()

		libtime.SleepString("24h")
	}
}

func (app *Application) PruneTSWatchersOnce() {
	for _, db := range app.DBConfig.TSWatchers.DBs {
		go func() {
			err := dal.NewTSWatcher(db).DeleteByDayInterval(nil, app.GeneralConfig.Watchers.DataRetention)
			if err != nil {
				logrus.WithFields(logrus.Fields{
					"Method":        "TSWatcher.DeleteByDayInterval",
					"DataRetention": app.GeneralConfig.Watchers.DataRetention,
				}).Error(err)
			}
		}()
	}
}

func (app *Application) PruneTSMetricsOnce() {
	for _, db := range app.DBConfig.TSMetrics.DBs {
		go func() {
			err := dal.NewTSMetric(db).DeleteByDayInterval(nil, app.GeneralConfig.Metrics.DataRetention)
			if err != nil {
				logrus.WithFields(logrus.Fields{
					"Method":        "TSMetric.DeleteByDayInterval",
					"DataRetention": app.GeneralConfig.Metrics.DataRetention,
				}).Error(err)
			}
		}()
	}
}

func (app *Application) WatchAll() {
	watcherRowsChan := make(chan []*dal.WatcherRow)

	// Fetch daemons and watchers data every 5 minutes
	go func() {
		for {
			daemonHostnames, _ := dal.NewDaemon(app.DBConfig.Core).AllHostnames(nil)
			groupedWatcherRows, _ := dal.NewWatcher(app.DBConfig.Core).AllSplitToDaemons(nil, daemonHostnames)
			watcherRowsChan <- groupedWatcherRows[app.Hostname]

			libtime.SleepString(app.GeneralConfig.Watchers.ListFetchInterval)
		}
	}()

	go func() {
		for {
			select {
			case watcherRows := <-watcherRowsChan:
				for _, watcherRow := range watcherRows {
					go func(watcherRow *dal.WatcherRow) {
						if watcherRow.IsPassive() {
							// Passive watching and triggering
							err := app.PassiveWatchOnce(watcherRow.ClusterID, watcherRow)
							if err != nil {
								logrus.WithFields(logrus.Fields{
									"Method":    "Application.PassiveWatchOnce",
									"ClusterID": watcherRow.ClusterID,
									"WatcherID": watcherRow.ID,
								}).Error(err)
							}

							err = app.RunTrigger(watcherRow.ClusterID, watcherRow)
							if err != nil {
								logrus.WithFields(logrus.Fields{
									"Method":    "Application.RunTrigger",
									"ClusterID": watcherRow.ClusterID,
									"WatcherID": watcherRow.ID,
								}).Error(err)
							}
						} else {
							// Active watching and triggering
							err := app.ActiveWatchOnce(watcherRow.ClusterID, watcherRow)
							if err != nil {
								logrus.WithFields(logrus.Fields{
									"Method":    "Application.ActiveWatchOnce",
									"ClusterID": watcherRow.ClusterID,
									"WatcherID": watcherRow.ID,
								}).Error(err)
							}
						}

						libtime.SleepString(watcherRow.CheckInterval)
					}(watcherRow)
				}
			}
		}
	}()
}

func (app *Application) PassiveWatchOnce(clusterID int64, watcherRow *dal.WatcherRow) error {
	affectedHosts, err := dal.NewHost(app.DBConfig.Core).AllByClusterIDQueryAndUpdatedInterval(nil, clusterID, watcherRow.SavedQuery.String, watcherRow.HostsLastUpdatedForPostgres())
	if err != nil {
		return err
	}

	numAffectedHosts := int64(len(affectedHosts))

	if numAffectedHosts == 0 || numAffectedHosts >= watcherRow.LowAffectedHosts {
		tsWatcherDataHosts := make([]string, numAffectedHosts)
		for i, affectedHost := range affectedHosts {
			tsWatcherDataHosts[i] = affectedHost.Name
		}

		tsWatcherData := make(map[string]interface{})
		tsWatcherData["hosts"] = tsWatcherDataHosts

		jsonData, err := json.Marshal(tsWatcherData)
		if err != nil {
			return err
		}

		// Write to ts_watchers asynchronously
		dbs := app.DBConfig.TSWatchers.PickMultipleForWrites()
		for _, db := range dbs {
			go func() {
				err := dal.NewTSWatcher(db).Create(nil, clusterID, watcherRow.ID, numAffectedHosts, jsonData)
				if err != nil {
					logrus.WithFields(logrus.Fields{
						"Method":          "TSWatcher.Create",
						"ClusterID":       clusterID,
						"WatcherID":       watcherRow.ID,
						"NumAffectedHost": numAffectedHosts,
					}).Error(err)
				}
			}()
		}
	}

	return err
}

func (app *Application) ActiveWatchOnce(clusterID int64, watcherRow *dal.WatcherRow) error {
	// Don't do anything if watcher is silenced.
	if watcherRow.IsSilenced {
		return nil
	}

	var hostnames []string

	// Checking hosts based on external lists.
	if len(watcherRow.HostsList()) > 0 {
		hostnames = watcherRow.HostsList()
	} else {
		hosts, err := dal.NewHost(app.DBConfig.Core).AllByClusterIDAndUpdatedInterval(nil, clusterID, watcherRow.HostsLastUpdatedForPostgres())
		if err != nil {
			return err
		}

		hostnames = make([]string, len(hosts))

		for i, host := range hosts {
			hostnames[i] = host.Name
		}
	}

	// If there are no hostnames, then we are not checking anything
	if len(hostnames) == 0 {
		return nil
	}

	numAffectedHosts := 0
	tsWatcherDataHosts := make([]string, 0)

	var errCollectorMutex sync.Mutex

	for _, hostname := range hostnames {
		if watcherRow.Command() == "ping" {
			go func(hostname string) {
				_, err := exec.Command("ping", "-c", "1", hostname).Output()
				if err != nil {
					errCollectorMutex.Lock()
					numAffectedHosts = numAffectedHosts + 1
					tsWatcherDataHosts = append(tsWatcherDataHosts, hostname)
					errCollectorMutex.Unlock()

					logrus.WithFields(logrus.Fields{
						"Method": "ping -c 1 " + hostname,
					}).Error(err)
				}
			}(hostname)

		} else if watcherRow.Command() == "ssh" {
			go func(hostname string) {
				sshOptions := []string{"-o BatchMode=yes", "-o ConnectTimeout=10"}

				if watcherRow.SSHPort() != "" {
					sshOptions = append(sshOptions, []string{"-p", watcherRow.SSHPort()}...)
				}

				userAtHost := hostname

				if watcherRow.SSHUser() != "" {
					userAtHost = fmt.Sprintf("%v@%v", watcherRow.SSHUser(), hostname)
				}

				sshOptions = append(sshOptions, userAtHost)

				outBytes, err := exec.Command("ssh", sshOptions...).CombinedOutput()
				outString := string(outBytes)

				// We only care about SSH connectivity
				if err != nil && !strings.Contains(outString, "Permission denied") && !strings.Contains(outString, "Host key verification failed") {
					errCollectorMutex.Lock()
					numAffectedHosts = numAffectedHosts + 1
					tsWatcherDataHosts = append(tsWatcherDataHosts, hostname)
					errCollectorMutex.Unlock()

					logrus.WithFields(logrus.Fields{
						"Method": "ssh -o BatchMode=yes -o ConnectTimeout=10 " + userAtHost,
					}).Error(err)
				}
			}(hostname)

		} else if watcherRow.Command() == "http" {
			go func(hostname string) {
				url := fmt.Sprintf("%v://%v:%s", watcherRow.HTTPScheme(), hostname, watcherRow.HTTPPort())

				client := &http.Client{}
				req, err := http.NewRequest(strings.ToUpper(watcherRow.HTTPMethod()), url, nil)
				if err != nil {
					logrus.WithFields(logrus.Fields{
						"URL":        url,
						"HTTPMethod": watcherRow.HTTPMethod(),
						"Method":     "http.NewRequest",
					}).Error(err)
					return
				}

				if watcherRow.HTTPUser() != "" || watcherRow.HTTPPass() != "" {
					req.SetBasicAuth(watcherRow.HTTPUser(), watcherRow.HTTPPass())
				}

				resp, err := client.Do(req)
				if err != nil {
					logrus.WithFields(logrus.Fields{
						"URL":        url,
						"HTTPMethod": watcherRow.HTTPMethod(),
						"Method":     "http.Client{}.Do",
					}).Error(err)
					return
				}

				if err != nil || (resp != nil && resp.StatusCode != watcherRow.HTTPCode()) {
					errCollectorMutex.Lock()
					numAffectedHosts = numAffectedHosts + 1
					tsWatcherDataHosts = append(tsWatcherDataHosts, hostname)
					errCollectorMutex.Unlock()

					logrus.WithFields(logrus.Fields{
						"URL":                url,
						"HTTPMethod":         watcherRow.HTTPMethod(),
						"StatusCode":         resp.StatusCode,
						"ExpectedStatusCode": watcherRow.HTTPCode(),
						"Method":             "HTTP Client fetch",
					}).Error(err)
				}
			}(hostname)
		}
	}

	if numAffectedHosts == 0 || int64(numAffectedHosts) >= watcherRow.LowAffectedHosts {
		tsWatcherData := make(map[string]interface{})
		tsWatcherData["hosts"] = tsWatcherDataHosts

		jsonData, err := json.Marshal(tsWatcherData)
		if err != nil {
			return err
		}

		// Write to ts_watchers asynchronously
		dbs := app.DBConfig.TSWatchers.PickMultipleForWrites()
		for _, db := range dbs {
			go func() {
				err := dal.NewTSWatcher(db).Create(nil, clusterID, watcherRow.ID, int64(numAffectedHosts), jsonData)
				if err != nil {
					logrus.WithFields(logrus.Fields{
						"Method":          "TSWatcher.Create",
						"ClusterID":       clusterID,
{'fraction_non_alphanumeric': 0.15596359572090052, 'fraction_numerical': 0.004087498004151365, 'mean_word_length': 4.436805555555556, 'pattern_counts': {'":': 61, '<': 7, '<?xml version=': 0, '>': 9, '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': '3484693', 'n_tokens_mistral': 12001, 'n_tokens_neox': 10408, 'n_words': 1590}