update logging

Author: Alextopher

api.go

@@ -5,3 +5,5 @@ import "github.com/gorilla/mux"
 func HandleAPI(r *mux.Router) {
 
 }
+
+// api/stats/{distro}

config.go

@@ -6,7 +6,7 @@ import (
 	"io/ioutil"
 	"log"
 
-	"github.com/COSI_Lab/Mirror/mirrorErrors"
+	"github.com/COSI_Lab/Mirror/logging"
 	"github.com/xeipuuv/gojsonschema"
 )
 
@@ -90,7 +90,7 @@ func getPassword(filename string) string {
 	bytes, err := ioutil.ReadFile(filename)
 
 	if err != nil {
-		mirrorErrors.Error("Could not read password file: "+filename, "WARN")
+		logging.Log(logging.Warn, "Could not read password file: ", filename, err.Error())
 	}
 
 	return string(bytes)

influx.go

@@ -1,16 +1,85 @@
 package main
 
 import (
+	"context"
+	"time"
+
+	"github.com/COSI_Lab/Mirror/logging"
 	influxdb2 "github.com/influxdata/influxdb-client-go/v2"
 	"github.com/influxdata/influxdb-client-go/v2/api"
 	"github.com/influxdata/influxdb-client-go/v2/api/write"
 )
 
 type DataPoint *write.Point
 
-func InfluxClients(token string) (api.WriteAPI, api.QueryAPI) {
+var writer api.WriteAPI
+var reader api.QueryAPI
+
+func InfluxClients(token string) {
 	// create new client with default option for server url authenticate by token
 	client := influxdb2.NewClient("https://mirror.clarkson.edu:8086", token)
 
-	return client.WriteAPI("COSI", "test"), client.QueryAPI("COSI")
+	writer = client.WriteAPI("COSI", "test")
+	reader = client.QueryAPI("COSI")
+}
+
+// Sends the latest NGINX stats to the database
+func SendTotalBytesByDistro(bytesByDistro map[string]int) {
+	// Measure time
+	t := time.Now()
+
+	// Create and send points
+	for short, bytes := range bytesByDistro {
+		p := influxdb2.NewPoint("mirror", map[string]string{"distro": short}, map[string]interface{}{"bytes_sent": bytes}, t)
+		writer.WritePoint(p)
+	}
+}
+
+// Loads the latest NGINX stats from the database
+// Returns a map of distro short names to total bytes sent and total in the map
+func QueryTotalBytesByDistro(shorts []string) (map[string]int, int) {
+	// Map from short names to bytes sent
+	bytesByDistro := make(map[string]int)
+
+	for i := 0; i < len(shorts); i++ {
+		bytesByDistro[shorts[i]] = 0
+	}
+
+	/*
+		from(bucket: \"test\")
+			|> range(start: -7d)
+			|> filter(fn: (r) => r[\"_measurement\"] == \"mirror\" and  r[\"_field\"] == \"bytes_sent\")
+			|> last()
+	*/
+	result, err := reader.Query(context.Background(), "from(bucket: \"test\") |> range(start: -7d) |> filter(fn: (r) => r[\"_measurement\"] == \"mirror\" and  r[\"_field\"] == \"bytes_sent\") |> last()")
+
+	if err != nil {
+		logging.Log(logging.Error, "Error querying influxdb", err)
+	}
+
+	total := 0
+	for result.Next() {
+		if result.Err() == nil {
+			distro, ok := result.Record().ValueByKey("distro").(string)
+			if !ok {
+				logging.Log(logging.Warn, "InitNGINXStats can not parse distro to string: ", distro)
+				continue
+			}
+
+			bytes, ok := result.Record().Value().(int64)
+			if !ok {
+				logging.Log(logging.Warn, "InitNGINXStats can not parse ", distro, " bytes to int ", distro+result.Record().String())
+				continue
+			}
+
+			if _, ok := bytesByDistro[distro]; ok {
+				bytesByDistro[distro] = int(bytes)
+				total += int(bytes)
+			}
+		} else {
+			logging.Log(logging.Warn, "InitNGINXStats Flux Query Error", result.Err())
+		}
+	}
+
+	return bytesByDistro, total
 }

mirrorErrors/errors.go renamed to logging/errors.go

@@ -1,27 +1,40 @@
-package mirrorErrors
+package logging
 
 import (
-	"bytes"
-	"encoding/json"
 	"errors"
-	"log"
-	"net/http"
+	"fmt"
 	"os"
+	"sync"
+	"time"
 )
 
 var rysncErrorCodes map[int]string
 var hookURL string
-var hookUnset = false
+
+var loggingLock sync.Mutex
+
+type MessageType int
+
+const (
+	// Info is the type for informational messages
+	Info MessageType = iota
+	// Warn is the type for warning messages
+	Warn
+	// Error is for when we lose funcitonality but it's fairly understood what went wrong
+	Error
+	// Panic is the type for fatal error messages and will print the stack trace
+	Panic
+	// Success is the type for successful messages
+	Success
+)
 
 func Setup() error {
 	hookURL = os.Getenv("HOOK_URL")
 	if hookURL == "" {
-		hookUnset = true
 		return errors.New("missing .env envirnment variable HOOK_URL, not interfacing with discord")
 	}
 
 	rysncErrorCodes = make(map[int]string)
-
 	rysncErrorCodes[0] = "Success"
 	rysncErrorCodes[1] = "Syntax or usage error"
 	rysncErrorCodes[2] = "Protocol incompatibility"
@@ -46,47 +59,24 @@ func Setup() error {
 	return nil
 }
 
-func sendHook(content string, url string) {
-	if !hookUnset {
-		values := map[string]string{"content": content}
-		json_data, err := json.Marshal(values)
-
-		if err != nil {
-			log.Fatal(err)
-		}
+func Log(messageType MessageType, v ...interface{}) {
+	loggingLock.Lock()
+	defer loggingLock.Unlock()
 
-		resp, err := http.Post(url, "application/json",
-			bytes.NewBuffer(json_data))
+	fmt.Print(time.Now().Format("2006/01/02 15:04:05 "))
 
-		if err != nil {
-			log.Fatal(err)
-		}
-
-		var res map[string]interface{}
-
-		json.NewDecoder(resp.Body).Decode(&res)
+	switch messageType {
+	case Info:
+		fmt.Print("\033[1m[INFO]    \033[0m| ")
+	case Warn:
+		fmt.Print("\033[1m\033[33m[WARN]   \033[0m| ")
+	case Error:
+		fmt.Print("\033[1m\033[31m[ERROR]   \033[0m| ")
+	case Panic:
+		fmt.Print("\033[1m\033[34m[PANIC]  \033[0m| ")
+	case Success:
+		fmt.Print("\033[1m\033[32m[SUCCESS] \033[0m| ")
 	}
 
-	// fmt.Println(res["json"])
-}
-
-func Error(message string, errorType string) {
-	// TODO: Have this handle logging to console and send hook
-	if errorType == "info" {
-		log.Printf("[INFO] %s", message)
-	} else if errorType == "warn" {
-		log.Printf("\033[33m[WARN]\033[0m %s", message)
-	} else if errorType == "error" {
-		log.Printf("\033[31m[ERROR]\033[0m %s", message)
-		sendHook(message, hookURL)
-	} else if errorType == "panic" {
-		log.Printf("[PANIC] %s", message)
-		sendHook(message, hookURL)
-	} else if errorType == "startup" {
-		log.Printf("[STARTUP] %s", message)
-		sendHook(message, hookURL)
-	} else {
-		log.Printf("\033[34m[DEBUG]\033[0m %s", message)
-		sendHook(message, hookURL)
-	}
+	fmt.Println(v...)
 }

main.go

@@ -4,21 +4,19 @@ import (
 	"os"
 	"time"
 
-	"github.com/COSI_Lab/Mirror/mirrorErrors"
+	"github.com/COSI_Lab/Mirror/logging"
 	"github.com/joho/godotenv"
 )
 
 func main() {
 	godotenv.Load()
 
 	// Setup error logger
-	err := mirrorErrors.Setup()
+	err := logging.Setup()
 	if err != nil {
-		mirrorErrors.Error(err.Error(), "error")
+		logging.Log(logging.Error, "Setting up logging", err)
 	}
 
-	mirrorErrors.Error("Starting Mirror", "startup")
-
 	// Load config file and check schema
 	config := ParseConfig("configs/mirrors.json", "configs/mirrors.schema.json")
 
@@ -27,8 +25,8 @@ func main() {
 		shorts = append(shorts, mirror.Name)
 	}
 
-	writer, reader := InfluxClients(os.Getenv("INFLUX_TOKEN"))
-	mirrorErrors.Error("Connected to InfluxDB", "startup")
+	InfluxClients(os.Getenv("INFLUX_TOKEN"))
+	logging.Log(logging.Success, "Connected to InfluxDB")
 
 	nginx_entries := make(chan *LogEntry, 100)
 	map_entries := make(chan *LogEntry, 100)
@@ -37,10 +35,10 @@ func main() {
 	// ReadLogs("/var/log/nginx/access.log", channels)
 
 	if os.Getenv("INFLUX_TOKEN") == "" {
-		mirrorErrors.Error("Missing .env envirnment variable INFLUX_TOKEN, not using database", "error")
+		logging.Log(logging.Error, "Missing .env envirnment variable INFLUX_TOKEN, not using database")
 	} else {
-		InitNGINXStats(shorts, reader)
-		go HandleNGINXStats(nginx_entries, writer)
+		InitNGINXStats(shorts)
+		go HandleNGINXStats(nginx_entries)
 	}
 
 	if InitWebserver() == nil {

map.go

@@ -3,13 +3,12 @@ package main
 import (
 	"encoding/binary"
 	"fmt"
-	"log"
 	"math"
 	"net"
 	"net/http"
 	"sync"
 
-	"github.com/COSI_Lab/Mirror/mirrorErrors"
+	"github.com/COSI_Lab/Mirror/logging"
 	"github.com/gorilla/mux"
 	"github.com/gorilla/websocket"
 	"github.com/thanhpk/randstr"
@@ -53,12 +52,10 @@ func serve(clients map[string]chan []byte, entries chan *LogEntry) {
 			prevSkip = skip
 
 			if skip {
-				// log.Printf("[INFO] MirrorMap no clients connected, skipping")
-				mirrorErrors.Error("MirrorMap no clients connected, skipping", "info")
+				logging.Log(logging.Info, "MirrorMap no clients connected, skipping")
 				continue
 			} else {
-				// log.Printf("[INFO] MirrorMap new clients connected, sending data")
-				mirrorErrors.Error("MirrorMap new clients connected, sending data", "info")
+				logging.Log(logging.Info, "MirrorMap new clients connected, sending data")
 			}
 		}
 
@@ -100,13 +97,12 @@ func socketHandler(w http.ResponseWriter, r *http.Request) {
 	// get the channel
 	ch := clients[id]
 
-	log.Printf("[INFO] Websocket new client connected %s : %s ", id, r.RemoteAddr)
-	// mirrorErrors.Error("Websocket new client connected", "info")
+	logging.Log(logging.Info, "Websocket new client connected", id, r.RemoteAddr)
 
 	// Upgrade our raw HTTP connection to a websocket based one
 	conn, err := upgrader.Upgrade(w, r, nil)
 	if err != nil {
-		mirrorErrors.Error("Error during connection upgradate", "warn")
+		logging.Log(logging.Warn, "Websocket upgrade failed", err)
 		return
 	}
 
@@ -123,7 +119,7 @@ func socketHandler(w http.ResponseWriter, r *http.Request) {
 	// Close connection gracefully
 	conn.Close()
 	clients_lock.Lock()
-	mirrorErrors.Error("Error sending message", "warn")
+	logging.Log(logging.Warn, "Error sending message", err, "disconnecting", id)
 	delete(clients, id)
 	clients_lock.Unlock()
 }
@@ -136,7 +132,8 @@ func registerHandler(w http.ResponseWriter, r *http.Request) {
 	clients_lock.Lock()
 	clients[id] = make(chan []byte, 10)
 	clients_lock.Unlock()
-	log.Printf("[INFO] Map new connection registered: %s\n", id)
+
+	logging.Log(logging.Info, "Map new connection registered", id)
 
 	// Send id to client
 	w.WriteHeader(200)

nginx.go

@@ -11,7 +11,7 @@ import (
 	"strings"
 	"time"
 
-	"github.com/COSI_Lab/Mirror/mirrorErrors"
+	"github.com/COSI_Lab/Mirror/logging"
 	"github.com/nxadm/tail"
 	"github.com/oschwald/geoip2-golang"
 )
@@ -40,7 +40,7 @@ var db *geoip2.Reader
 func InitDb() (err error) {
 	db, err = geoip2.Open("GeoLite2-City.mmdb")
 	if err != nil {
-		mirrorErrors.Error("could not open geolite city db", "error")
+		logging.Log(logging.Error, "Could not open geolite city db")
 		return err
 	}
 
@@ -60,13 +60,13 @@ func InitRegex() (err error) {
 func ReadLogFile(logFile string, ch1 chan *LogEntry, ch2 chan *LogEntry) (err error) {
 	if reQuotes == nil {
 		if InitRegex() != nil {
-			mirrorErrors.Error("could not compile nginx log parsing regex", "error")
+			logging.Log(logging.Error, "could not compile nginx log parsing regex")
 		}
 	}
 
 	if db == nil {
 		if InitDb() != nil {
-			mirrorErrors.Error("could not initilze geolite city db", "error")
+			logging.Log(logging.Error, "could not initilze geolite city db")
 		}
 	}
 
@@ -99,13 +99,13 @@ func ReadLogFile(logFile string, ch1 chan *LogEntry, ch2 chan *LogEntry) (err er
 func ReadLogs(logFile string, ch1 chan *LogEntry, ch2 chan *LogEntry) (err error) {
 	if reQuotes == nil {
 		if InitRegex() != nil {
-			mirrorErrors.Error("could not compile nginx log parsing regex", "error")
+			logging.Log(logging.Error, "could not compile nginx log parsing regex")
 		}
 	}
 
 	if db == nil {
 		if InitDb() != nil {
-			mirrorErrors.Error("could not initilze geolite city db", "error")
+			logging.Log(logging.Error, "could not initilze geolite city db")
 		}
 	}
 
@@ -128,7 +128,7 @@ func ReadLogs(logFile string, ch1 chan *LogEntry, ch2 chan *LogEntry) (err error
 		}
 	}
 
-	mirrorErrors.Error("Closing ReadLogs *LogEntry channel for unknown reason. This should not happen!", "error")
+	logging.Log(logging.Panic, "Closing ReadLogs for unknown reason.")
 	close(ch1)
 	close(ch2)